ActiveMQ的简单使用说明。

一.下载安装

前提：activemq需要依赖jre，所以提前下载安装配置好jdk。

1. 访问官网activemq，选择对应系统进行下载(这里以Linux举例)
   

2. 解压bin.tar.gz包

tar -xzvf activemq.bin.tar.gz
1

3. 启动

cd apache-activemq-5.7.0/bin
./activemq start
1
2

4. 查看启动状态

./activemq status
1

出现如下ActiveMQ is running关键词则说明已经启动成功

默认启动端口有两个，一个是web控制台端口8161，另一个是服务端口61616，并且服务连接的默认账号密码均为admin
web控制台访问界面：localhost:8161

二.集成SpringBoot

我们通过一个简单的生产/消费关系，建立mq的生产者和消费者加深印象

1. 相关pom依赖

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-activemqartifactId>
dependency>

<dependency>
    <groupId>org.apache.activemqgroupId>
    <artifactId>activemq-poolartifactId>
    <version>5.12.1version>
dependency>


<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-webartifactId>
dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2. 配置文件

spring:
  activemq:
    #activeMQ的ip和端口号
    broker-url: tcp://192.162.20.87:61616
    #activeMq账号
    user: admin
    #activeMq密码
    password: admin
    # 队列名-支持自定义
    queue-name: test
    # 主题名-支持自定义
    topic-name: test
    pool:
      # 连接池启动
      enabled: true
      # 最大连接数
      max-connections: 10
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3. 消费者监听配置类
   activemq比较特殊的一点是，它区分两种消费模式，对应也是两种生产模式，分别是queue和topic。

@EnableJms
@Configuration
public class ActiveMqConfig {

    @Value("${spring.activemq.broker-url}")
    private String brokerUrl;

    @Value("${spring.activemq.user}")
    private String userName;

    @Value("${spring.activemq.password}")
    private String password;

    @Value("${spring.activemq.queue-name}")
    private String queueName;

    @Value("${spring.activemq.topic-name}")
    private String topicName;

    @Bean(name = "queue")
    public Queue queue() {
        return new ActiveMQQueue(queueName);
    }

    @Bean(name = "topic")
    public Topic topic(){
        return new ActiveMQTopic(topicName);
    }

    @Bean
    public ConnectionFactory connectionFactory(){
        return new ActiveMQConnectionFactory(userName, password, brokerUrl);
    }

    /**
     * 在Queue模式中，对消息的监听需要对containerFactory进行配置
     * @param connectionFactory
     * @return
     */
    @Bean("queueListener")
    public JmsListenerContainerFactory<?> queueJmsListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(false);
        return factory;
    }

    /**
     * 在topic模式中，对消息的监听需要对containerFactory进行配置
     * @param connectionFactory
     * @return
     */
    @Bean("topicListener")
    public JmsListenerContainerFactory<?> topicJmsListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true);
        return factory;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

4. 消费者

@Component
public class ReportConsumerListener {
	/**
     * queue模式的消费者
     * @param msg
     */
    @JmsListener(destination="${spring.activemq.queue-name}", containerFactory="queueListener")
    public void queueConsumer(String msg) {
        System.out.println("queue收到消息：" + msg);
    }

    /**
     * topic模式的消费者
     * @param msg
     */
    @JmsListener(destination = "${spring.activemq.topic-name}", containerFactory = "topicListener")
    public void topicConsumer(String msg){
        System.out.println("topic收到消息：" + msg);
    }
	
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

5. 生产者
   为了模拟消费的情况，我们创建一组对应queue和topic模式的生产者来模拟生产

@RestController
@RequestMapping("/activemq")
public class ProducerController {

    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    @Autowired
    private Queue queue;

    @Autowired
    private Topic topic;

    /**
     * queue模式mq消息发送
     * @param msg
     * @return
     */
    @GetMapping("/queue")
    public String sendQueue(String msg) {
        jmsMessagingTemplate.convertAndSend(queue, msg);
        return "success, 发送消息：" + msg;
    }

    /**
     * topic模式一对多的消息队列的生产者
     * @param msg
     */
    @GetMapping("/topic")
    public String sendMsgTopic(@RequestParam String msg){
        jmsMessagingTemplate.convertAndSend(topic,msg);
        return "success, 发送消息：" + msg;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

• 测试queue模式：
  curl -X “GET” “http://localhost:8080/activemq/queue?msg=测试消息”
  控制台打印：queue收到消息：测试消息

• 测试topic模式：
  curl -X “GET” “http://localhost:8080/activemq/topic?msg=测试消息”
  控制台打印：topic收到消息：测试消息

这个时候我们访问activemq控制台localhost:8161，在对应的topic/queue标签页下也能找到对应主题名的消费情况；

参考资料：

• SpringBoot整合ActiveMQ实现Queue和Topic两种模式