1、定义
2、消息模型
3、安装activeMQ
4、Spring Boot集成ActiveMQ
- 4.1、点对点模式
基于内存的ActiveMQ
不使用连接池，如果使用连接池还需在pom中添加activemq-pool的依赖
独立安装的ActiveMQ
spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.user=admin
spring.activemq.password=admin
- 4.2、发布订阅模式
- 4.3、同时支持两种形式
5、其他注意事项

1、定义

ActiveMQ是遵从 JMS （即java的消息服务，JMS的客户端之间可以通过JMS服务进行异步的消息传输）规范的消息中间件，

2、消息模型

2.1、点对点

Point-to-Point(P2P)，点对点
P2P模式图

在点对点模型中，一般消息由发送者将消息发送到消息队列中，然后，接收者从消息队列中消费消息，消息被消费者消费之后，消息就不存在了。

2.2、发布订阅模型

Publish/Subscribe(Pub/Sub)，发布订阅模型
Pub/Sub模式图

在发布订阅模型中，发布者通常将消息发布到主题（topic）中，然后，订阅者通过订阅主题来消费消息，与 P2P 模型不同的是，发布订阅模型的消息是可以被多次消费的！

2.3、两种模式的区别

P2P在发送者和接收者之间没有时间上的依赖性，也就是说发送者发送了消息之后，不管接收者有没有运行，不会影响消息发送到队列，而Pub/Sub模式有时间上的依赖性，消费者必须先订阅主题，才能够消费消息。
P2P模式的每个消息只能有一个消费者，消费完了消息就不存在了，Pub/Sub模式可以有多个消费者。

3、安装activeMQ

包括linux和Windows的不同版本地址

3.1、windows

里面有一个 activemq 的可执行文件，打开 cmd，执行：

启动命令: activemq start
关闭： activemq stop

3.2、linux

解压到指定目录
```
sudo tar zxvf activemq-x.x.x-bin.tar.gz
```
进入到 bin 目录，执行下面命令
```
./activemq start
```
关闭
```
./activemq stop
```

3.3、后台管理界面

启动成功之后，可以输出http://localhost:8161/admin/查看 ActiveMQ 的后台管理界面，用户名和密码都为 admin

4、Spring Boot集成ActiveMQ

4.1、点对点模式

引入依赖

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

此时如果不需要web或其他相关处理，只引入该依赖即可。如果使用pool的话, 就需要在pom中加入以下依赖：
```
<dependency>
   <groupId>org.apache.activemq</groupId>
   <artifactId>activemq-pool</artifactId>
</dependency>
```
application.properties配置 ```java

基于内存的ActiveMQ
spring.activemq.in-memory=true

不使用连接池，如果使用连接池还需在pom中添加activemq-pool的依赖
spring.activemq.pool.enabled=false

独立安装的ActiveMQ

spring.activemq.broker-url=tcp://127.0.0.1:61616

spring.activemq.user=admin

spring.activemq.password=admin


- 队列模式实例
首先，我们来实现基于队列（Queue）形式的实现。这里需要用到两个类ActiveMQQueue和JmsMessagingTemplate。前者是由ActiveMQ对javax.jms.Queue的接口实现。后者为Spring提供发送消息的工具类，结合Queue对消息进行发送。<br />JmsMessagingTemplate默认已经被实例化，直接拿来使用即可。而ActiveMQQueue则需要我们进行实例化，并传入消息队列的名称。
```java
@Configuration
public class MyMqConfig {
    @Bean
    public Queue queue() {
        return new ActiveMQQueue("sms.queue");
    }
}

生产者对应代码如下： ```java @Component public class Producer {

@Resource private JmsMessagingTemplate jmsMessagingTemplate;

@Resource private Queue queue;

public void sendMsg(String msg) {
```
  System.out.println("发送消息内容 :" + msg);
  this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
```
}

}


- 消费者对应的配置如下：
```java
@Component
public class Consumer {
    @JmsListener(destination = "sms.queue")
    public void receiveMsg(String text) {
        System.out.println("接收到消息 : "+text);
    }
}

Spring提供了注解式监听器端点：使用@JmsListener。使用@JmsListener托管bean的带注释方法对其进行订阅。在Java8中，@JmsListener是一个可重复的注解，可以关联多个JMS destinations到同一个方法中。而在Java 6和7中，可以使用@JmsListeners注解。
其中destination指定监控的消息队列名称为“sms.queue”。当队列sms.queue中有消息发送时会触发此方法的执行，text为消息内容。

单元测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class ActiveMqTests {
  @Autowired
  private Producer producer;
  @Test
  public void sendSimpleQueueMessage() {
      this.producer.sendMsg("提现200.00元");
  }
}

打印如下信息：
```
发送消息内容 :提现200.00元
接收到消息 : 提现200.00元
```
说明消息可以正常发送和接收。如果是基于内存模式，在执行单元测试时会打印出“javax.jms.JMSException: peer (vm://localhost#1) stopped.”异常日志，这是Info级别的错误，是ActiveMQ的一个bug。

4.2、发布订阅模式

广播发送的消息，可以被多个消费者接收。这里我们就在原有的基础上进行广播消息的添加。
首先，Spring Boot集成ActiveMQ时默认只支持队列或者广播之一，通过配置项spring.jms.pub-sub-domain来指定，true 为广播模式，false为队列模式，默认情况下支持队列模式。
此时要使用广播模式，则需在配置文件中添加如下配置：

spring.jms.pub-sub-domain=true

需要注意的是，此时队列模式不可正常工作。
然后在MyMqConfig中添加：

@Bean
public Topic topic() {
    return new ActiveMQTopic("sms.topic");
}

这里创建了ActiveMQTopic，并将topic的名称指定为sms.topic。
Producer中新增如下代码：

@Resource
private Topic topic;
public void sendTopic(String msg) {
    System.out.println("发送Topic消息内容 :"+msg);
    this.jmsMessagingTemplate.convertAndSend(this.topic, msg);
}

为了演示多个广播接收者，在Comsumer中新增两个消费者：

@JmsListener(destination = "sms.topic")
public void receiveTopic1(String text) {
    System.out.println("receiveTopic1接收到Topic消息 : " + text);
}
@JmsListener(destination = "sms.topic")
public void receiveTopic2(String text) {
    System.out.println("receiveTopic2接收到Topic消息 : " + text);
}

单元测试类中新增如下测试：

@Test
public void sendSimpleTopicMessage() {
    this.producer.sendTopic("提现200.00元");
}

此时，执行单元测试，便可看到如下日志信息：

发送Topic消息内容 :提现200.00元
receiveTopic2接收到Topic消息 : 提现200.00元
receiveTopic1接收到Topic消息 : 提现200.00元

4.3、同时支持两种形式

在上面的实例中，要么支持队列模式要么支持广播模式，如果在生产环境中两者都需要支持，那么就需要自定义JmsListenerContainerFactory实例。当然，如果Spring Boot默认的配置无法满足需求，也可以自定义该类，这里只是其中场景之一。
基本配置和使用步骤：通过DefaultJmsListenerContainerFactory创建自定义的JmsListenerContainerFactory实例，在@JmsListener注解中通过containerFactory属性进行引用。
在MyMqConfig配置类中新增如下配置：

@Bean("queueListenerFactory")
public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setPubSubDomain(false);
    return factory;
}
@Bean("topicListenerFactory")
public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    //设置为发布订阅方式, 默认情况下使用的生产消费者方式
    factory.setPubSubDomain(true);
    return factory;
}

这里分别实例化了基于队列和订阅的工厂类。然后分别在对应的消费者方法上添加containerFactory属性。示例代码如下：

@JmsListener(destination = "sms.queue", containerFactory = "queueListenerFactory")
public void receiveMsg(String text) {
    System.out.println("接收到消息 : " + text);
}
@JmsListener(destination = "sms.topic", containerFactory = "topicListenerFactory")
public void receiveTopic1(String text) {
    System.out.println("receiveTopic1接收到Topic消息 : " + text);
}

分别执行两种形式的消息，发现都正常互利。同时，此时配置文件中的项spring.jms.pub-sub-domain也无效了。

5、其他注意事项

activeMq的端口号是61616；
使用topic,需要配置spring.jms.pub-sub-domain=true;
queue如果没有消费者，会将信息存储到queue中；
发送的消息为对象的时候，需要将对象序列化；消费者接收对象信息时需要使用ObjectMessage进行转化；
使用JmsListener注解中的containerFactory属性，可以配置spring.jms.pub-sub属性，实现同时接收queque和topic；
queue为点对点模式；tipic为发布订阅模式；
示例中的消息队列名称（sms.queue和sms.topic）可根据需要设置成配置属性；

源码地址

ActiveMQ