张家港个人网站制作,恩施网页定制,最好的包装设计公司哪家好,代码做网站一、消息队列消息队列中间件是分布式系统中重要的组件#xff0c;主要解决应用耦合、异步消息、流量削锋等问题#xff0c;实现高性能、高可用、可伸缩和最终一致性架构#xff0c;是大型分布式系统不可缺少的中间件。目前在生产环境中使用较多的消息队列有ActiveMQ、Rabbit…一、消息队列消息队列中间件是分布式系统中重要的组件主要解决应用耦合、异步消息、流量削锋等问题实现高性能、高可用、可伸缩和最终一致性架构是大型分布式系统不可缺少的中间件。目前在生产环境中使用较多的消息队列有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/admin2、安全配置安装完成ActiveMQ后任何连接到ActiveMQ的程序都可以创建和消费队列可以通过修改配置文件conf/activemq.xml来加入身份验证在文件的borker标签中加入plugins simpleAuthenticationPlugin users authenticationUser usernameadmin passwordadmin groupsusers,admins/ /users /simpleAuthenticationPlugin
/plugins控制台账号密码修改conf/jetty.xml确保authenticate的值是true。bean idsecurityConstraint classorg.eclipse.jetty.util.security.Constraint property namename valueBASIC / property nameroles valueadmin / property nameauthenticate valuetrue /
/bean登陆管控台的帐号和密码在conf/jetty-realm.properties文件。# Defines users that can access the web (console, demo, etc.)
# username: password [,rolename ...]
admin: admin, admin重启生效。A、相关依赖dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-activemq/artifactId
/dependencyB、配置文件在使用ActiveMQ时有两种使用方式一种是使用独立安装的ActiveMQ在生产环境推荐使用这种另一种是使用基于内存ActiveMQ 在调试阶段建议使用这种方式。# 基于内存
ActiveMQspring.activemq.in-memorytrue# 不适应连接池
spring.activemq.pool.enabledfalse# 独立安装ActiveMQ
#spring.activemq.broker-urltcp://10.255.242.168:61616
#spring.activemq.useradmin
#spring.activemq.passwordadmin三、队列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、创建TopicConfiguration
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和广播TopicSpring 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-memorytrue
#spring.activemq.pool.enabledfalse# 独立安装的ActiveMQ
spring.activemq.broker-urltcp://10.255.242.168:61616
spring.activemq.useradmin
spring.activemq.passwordadmin# 结束之前等待的时间
#spring.activemq.close-timeout15s# 等待消息发送响应的时间设置为0永远等待
spring.activemq.send-timeout0# 默认情况下ActiveMQ提供的是queue模式若要使用topic模式需要配置下面配置
#spring.jms.pub-sub-domaintrue#账号
# spring.activemq.useradmin# 密码
# spring.activemq.passwordadmin# 是否信任所有包
#spring.activemq.packages.trust-all# 要信任的特定包逗号分隔
#spring.activemq.packages.trusted# 当连接请求和满时是否阻塞设置false会抛出JMSException异常
#spring.activemq.pool.block-if-fulltrue# 如果池满则在抛出异常前阻塞时间
#spring.activemq.pool.block-if-full-timeout-1ms# 是否在启动时创建连接可以在启动时用于热加载
#spring.activemq.pool.create-connection-on-startuptrue# 是否用Pooledconnectionfactory代替普通的ConnectionFactory
#spring.activemq.pool.enabledfalse# 连接过期超时
#spring.activemq.pool.expiry-timeout0ms# 连接空闲超时
#spring.activemq.pool.idle-timeout30s# 连接池最大连接数
#spring.activemq.pool.max-connections1# 每个连接的有效会话的最大数目。
#spring.activemq.pool.maximum-active-session-per-connection500# 当有JMSException时尝试重新连接
#spring.activemq.pool.reconnect-on-exceptiontrue# 空闲连接清除线程之间运行的时间当为负数时没有空闲连接驱逐线程运行
#spring.activemq.pool.time-between-expiration-check-1ms# 是否只使用一个MessageProducer
#spring.activemq.pool.use-anonymous-producerstrue
六、总结消息中间件广泛应用在大型互联网架构中利用消息中间件队列和广播各自的特性可以支持很多业务比如群发发送短信、给单个用户发送邮件等。ActiveMQ是一款非常流行的消息中间件它的特点是部署简单、使用方便比较适合中小型团队。Spring Boot提供了集成ActiveMQ对应的组件在Spring Boot中使用ActiveMQ只需要添加相关注解即可。
