启动activemq_「Java」 - SpringBoot amp; ActiveMQ

一、消息队列

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合、异步消息、流量削锋等问题，实现高性能、高可用、可伸缩和最终一致性架构，是大型分布式系统不可缺少的中间件。

目前在生产环境中使用较多的消息队列有ActiveMQ、RabbitMQ、Kafka、RocketMQ等。

A、特性

异步性：将耗时的同步操作以消息的方式进行异步化处理，减少了同步等待的时间；
松耦合：消息队列减少了服务之间的耦合性，不同的服务可以通过消息队列进行通信，而不用关心彼此的实现细节，只要定义好消息的格式就行；
分布式：通过对消费者的横向扩展，降低了消息队列阻塞的风险，以及单个消费者产生单点故障的可能性；
可靠性：消息队列一般会把接收到的消息存储到本地硬盘上，这样即使应用挂掉或者消息队列本身挂掉，消息也能够重新加载。

B、JMS规范

JMS即Java消息服务（Java Message Service）应用程序接口，是Java面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

JMS的消息机制有2种模型，一种是Point to Point，表现为队列的形式，发送的消息，只能被一个接收者取走；另一种是Topic，可以被多个订阅者订阅，类似于群发。

ActiveMQ就是JMS的一个实现。

二、ActiveMQ介绍

ActiveMQ是Apache软件基金下的一个开源软件，它遵循JMS 1.1规范，是消息驱动中间件软件。它为企业消息传递提供高可用、出色性能、可扩展、稳定和安全保障。ActiveMQ使用Apache许可协议，因此，任何人都可以使用和修改它而不必反馈任何改变。

ActiveMQ的目标是在尽可能多的平台和语言上提供一个标准的，消息驱动的应用集成。ActiveMQ实现JMS规范并在此之上提供大量额外的特性。ActiveMQ支持队列和订阅两种模式的消息发送。

Spring Boot提供了ActiveMQ组件spring-boot-starter-activemq，用来支持ActiveMQ在Spring Boot体系内使用。

A、ActiveMQ安装

1、下载安装启动

# 安装JDK并配置环境# 下载activemq
wget http://archive.apache.org/dist/activemq/5.12.2/apache-activemq-5.12.2-bin.tar.gz# 解压安装
cd /usr/local/apache-activemq-5.12.2/bin# 启动ActiveMQ
./activemq start# web控制台
http://192.168.240.131:8161
admin/admin

2、安全配置

安装完成ActiveMQ后，任何连接到ActiveMQ的程序都可以创建和消费队列，可以通过修改配置文件conf/activemq.xml来加入身份验证，在文件的borker标签中加入：

<plugins> <simpleAuthenticationPlugin> <users> <authenticationUser username="admin" password="admin" groups="users,admins"/> </users> </simpleAuthenticationPlugin> 
</plugins>

控制台账号密码，修改conf/jetty.xml，确保authenticate的值是true。

<bean id="securityConstraint" class="org.eclipse.jetty.util.security.Constraint"> <property name="name" value="BASIC" /> <property name="roles" value="admin" /> <property name="authenticate" value="true" /> 
</bean>

登陆管控台的帐号和密码在conf/jetty-realm.properties文件。

# Defines users that can access the web (console, demo, etc.)
# username: password [,rolename ...]
admin: admin, admin

重启生效。

A、相关依赖

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

B、配置文件

在使用ActiveMQ时有两种使用方式，一种是使用独立安装的ActiveMQ，在生产环境推荐使用这种；另一种是使用基于内存ActiveMQ ，在调试阶段建议使用这种方式。

# 基于内存 
ActiveMQspring.activemq.in-memory=true# 不适应连接池
spring.activemq.pool.enabled=false# 独立安装ActiveMQ
#spring.activemq.broker-url=tcp://10.255.242.168:61616
#spring.activemq.user=admin
#spring.activemq.password=admin

三、队列（Queue）

队列发送的消息，只能被一个消费者接收。

A、创建队列

@Configuration
public class ActiveMqConfig
{@Beanpublic Queue queue(){return new ActiveMQQueue("isisiwish.test.queue");}
}

使用定义了队列queue命名为isisiwish.test.queue。

B、消息生产者

@Slf4j
@Component
public class Producer
{@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Autowiredprivate Queue queue;public void sendQueue(String msg){log.info("send queue msg : {}", msg);this.jmsMessagingTemplate.convertAndSend(this.queue, msg);}
}

JmsMessagingTemplate是Spring提供发送消息的工具类，使用JmsMessagingTemplate和创建好的queue对消息进行发送。

C、消息消费者

@Slf4j
@Component
public class ConsumerA
{@JmsListener(destination = "isisiwish.test.queue")public void receiveQueue(String text){log.info("ConsumerA queue msg : {}", text);}
}@Slf4j
@Component
public class ConsumerB
{@JmsListener(destination = "isisiwish.test.queue")public void receiveQueue(String text){log.info("ConsumerB queue msg : {}", text);}
}

使用注解@JmsListener(destination = "isisiwish.test.queue")，表示此方法监控了名为isisiwish.test.queue的队列。当队列isisiwish.test.queue中有消息发送时会触发此方法的执行，text为消息内容。

D、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class MqActivemqQueueApplicationTests
{@Autowiredprivate Producer producer;@Testpublic void sendSimpleQueueMessage() throws InterruptedException{this.producer.sendQueue("Test queue message");}@Testpublic void send100QueueMessage() throws InterruptedException{for (int i = 0; i < 100; i++){this.producer.sendQueue("Test queue message " + i);}Thread.sleep(1000L);}
}

当有多个消费者监听一个队列时，消费者会自动均衡负载的接收消息，并且每个消息只能有一个消费者所接收。

PS：控制台输出javax.jms.JMSException: peer (vm://localhost#1) stopped.报错信息可以忽略。

四、广播（Topic）

广播发送的消息，可以被多个消费者接收。

A、创建Topic

@Configuration
public class ActiveMqConfig
{@Beanpublic Topic topic(){return new ActiveMQTopic("isisiwish.test.topic");}
}

B、消息生产者

@Slf4j
@Component
public class ConsumerA
{@JmsListener(destination = "isisiwish.test.topic")public void receiveTopic(String text){log.info("ConsumerA topic msg : {}", text);}
}@Slf4j
@Component
public class ConsumerB
{@JmsListener(destination = "isisiwish.test.topic")public void receiveTopic(String text){log.info("ConsumerB topic msg : {}", text);}
}

C、消息消费者

@Slf4j
@Component
public class Producer
{@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Autowiredprivate Topic topic;public void sendTopic(String msg){log.info("send queue msg : {}", msg);this.jmsMessagingTemplate.convertAndSend(this.topic, msg);}
}

D、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class MqActivemqTopicApplicationTests
{@Autowiredprivate Producer producer;@Testpublic void sendSimpleTopicMessage() throws InterruptedException{this.producer.sendTopic("Test Topic message");Thread.sleep(1000L);}
}

广播（Topic）是一个发送者多个消费者的模式，两个消费者都收到了发送的消息。

五、同时支持队列（Queue）和广播（Topic）

Spring Boot集成ActiveMQ的项目默认只支持队列或者广播中的一种，通过配置项 spring.jms.pub-sub-domain的值来控制，true为广播模式，false为队列模式，默认情况下支持队列模式。

如果需要在同一项目中既支持队列模式也支持广播模式，可以通过DefaultJmsListenerContainerFactory创建自定义的JmsListenerContainerFactory实例，之后在@JmsListener注解中通过containerFactory属性引用它。

分别创建两个自定义的JmsListenerContainerFactory实例，通过pubSubDomain来控制是支持队列模式还是广播模式。

@Configuration
@EnableJms
public class ActiveMqConfig
{@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;}@Beanpublic Queue queue(){return new ActiveMQQueue("isisiwish.test.queue");}@Beanpublic Topic topic(){return new ActiveMQTopic("isisiwish.test.topic");}
}

在消费者接收的方法中，指明使用containerFactory接收消息。

@Slf4j
@Component
public class ConsumerA
{@JmsListener(destination = "isisiwish.test.queue", containerFactory = "queueListenerFactory")public void receiveQueue(String text){log.info("ConsumerA queue msg : {}", text);}@JmsListener(destination = "isisiwish.test.topic", containerFactory = "topicListenerFactory")public void receiveTopic(String text){log.info("ConsumerA topic msg : {}", text);}
}

常用配置。

# 基于内存的ActiveMQ
#spring.activemq.in-memory=true
#spring.activemq.pool.enabled=false# 独立安装的ActiveMQ
spring.activemq.broker-url=tcp://10.255.242.168:61616
spring.activemq.user=admin
spring.activemq.password=admin# 结束之前等待的时间
#spring.activemq.close-timeout=15s# 等待消息发送响应的时间，设置为0永远等待
spring.activemq.send-timeout=0# 默认情况下ActiveMQ提供的是queue模式，若要使用topic模式需要配置下面配置
#spring.jms.pub-sub-domain=true#账号
# spring.activemq.user=admin# 密码
# spring.activemq.password=admin# 是否信任所有包
#spring.activemq.packages.trust-all=# 要信任的特定包（逗号分隔）
#spring.activemq.packages.trusted=# 当连接请求和满时是否阻塞，设置false会抛出JMSException异常
#spring.activemq.pool.block-if-full=true# 如果池满，则在抛出异常前阻塞时间
#spring.activemq.pool.block-if-full-timeout=-1ms# 是否在启动时创建连接，可以在启动时用于热加载
#spring.activemq.pool.create-connection-on-startup=true# 是否用Pooledconnectionfactory代替普通的ConnectionFactory
#spring.activemq.pool.enabled=false# 连接过期超时
#spring.activemq.pool.expiry-timeout=0ms# 连接空闲超时
#spring.activemq.pool.idle-timeout=30s# 连接池最大连接数
#spring.activemq.pool.max-connections=1# 每个连接的有效会话的最大数目。
#spring.activemq.pool.maximum-active-session-per-connection=500# 当有JMSException时尝试重新连接
#spring.activemq.pool.reconnect-on-exception=true# 空闲连接清除线程之间运行的时间，当为负数时，没有空闲连接驱逐线程运行
#spring.activemq.pool.time-between-expiration-check=-1ms# 是否只使用一个MessageProducer
#spring.activemq.pool.use-anonymous-producers=true