一、序言
demo2 留下了两个问题:
1.我们利用demo2 的配置,在queue 模式下 连续发送10W消息出现出现状况。
2.topic 模式下,消费者重启时间段收不到监听的信息怎么办?
二、问题解析:
1.测试 发送10W消息,中途会出现
socket: tcp://localhost:61616 due to: java.net.BindException: Address already in use: JVM_Bind 异常。
你关掉activemq,利用netstat -aon | findstr "61616" 发现没有这个端口占用情况,查阅资料才知道:
http://activemq.apache.org/jmstemplate-gotchas.html
里面解释道,发送消息的时候创建connection,session 还要关闭,比较费资源,我猜测当创建销毁操作没测试完成的时候,另一个消息发送的时候,发现端口被占着,就会出现这个种情况,也就是说当发送频率比较高的情况,容易出现,文档建议用 pool 的东西。
一共有spring 的CachingConnectionFactory 和 activemq 的PooledConnectionFactory,由于PooledConnectionFactory 这东西要 activemq-pool.jar ,因此我还是选择CachingConnectionFactory~。~。
配置spring-jms.xml更改如下:
<!-- jms 连接工厂 --> <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616?jms.useAsyncSend=false" /> </bean> <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="connectionFactory"/> <!-- Session缓存数量,这里属性也可以直接在这里配置 --> <property name="sessionCacheSize" value="100" /> <!-- 基本的bean模板 --> <bean id = "jmsTemplate" class = "org.springframework.jms.core.JmsTemplate"> <!-- 链接工厂,这里应用缓存池 就行了--> <property name="connectionFactory" ref="cachingConnectionFactory"/> </bean> </bean>
这样操作了时候,发送20W条信息也没出现过问题,而且速度比发送快了几十倍......
2.topic 模式下如果A 消费者挂了,就收不到消息了,而我又想它收到消息,我们先来尝试持久化吧!
1.对于持久化,并不默认,其实queue 默认就持久化在文件里面的,但是topic 模式下我们得开启持久化配置, 在activemq.xml 里面有这样的配置:
<!-- 这里设置true 就算开启了 --> <broker xmlns="http://activemq.apache.org/schema/core" persistent="true" brokerName="localhost" dataDirectory="${activemq.base}/data"> <!-- 这里是文件存放的位置,其他的说明暂时不讲 --> <persistenceAdapter> <amqPersistenceAdapter syncOnWrite="true" directory="${activemq.base}/data2" maxFileLength="3mb"/> </persistenceAdapter>
2.持久化我们在spring-jms.xml 里面还得开启几个东西:
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd"> <!-- jms 连接工厂 --> <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616?jms.useAsyncSend=true" /> </bean> <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="connectionFactory"/> <!-- Session缓存数量 --> <property name="sessionCacheSize" value="100" /> </bean> <!-- 基本的bean模板 --> <bean id = "jmsTemplate" class = "org.springframework.jms.core.JmsTemplate"> <!-- 链接工长 --> <property name="connectionFactory" ref="cachingConnectionFactory"/> <!-- 进行持久化 --> <property name="deliveryMode" value="2" /> <!--订阅 发布模式 --> <property name="pubSubDomain" value="true" /> </bean> <!-- 消息订阅模式 --> <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> <!-- 订阅消息的名字 --> <constructor-arg index="0" value="orderTopic"/> </bean> </beans>
3.想想我们订阅者需要做些什么呢?发布者发布消息,订阅者去消费,这是1对多的形式,我们可以这样理解:公司设定很多活动代金卷,去参加活动的人都能领取,当然这分两种情况,第一种 就是我们前面测试的,只要我公司门口等(监听),活动开始(发布)就能领取了,如果你当时没在,就领取不到。第二种:很多情况下,公司搞活动我们不会等在那里,只要活动开始了,那么我过段时间也可以去,礼品公司会保留的,这种情况会导致多次领取,因此总要登记一下嘛,不能你领取了,过一会又来吧?activemq 里面会有clientId 标示来区分,类似于身份证ID嘛。
当然有些情况下, 我们一个ID 可以领取多个不同的奖品,因此还得需要个字段标示:durableSubscriptionName,标示我们领取哪个礼品,下面先看配置
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd"> <!-- jms 连接工厂 --> <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616?jms.useAsyncSend=true" /> </bean> <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="connectionFactory"/> <!-- 接收者ID --> <property name="clientId" value="clientA" /> </bean> <!-- 消息订阅模式 --> <bean id="topicCustomerA" class="org.apache.activemq.command.ActiveMQTopic"> <!-- 订阅消息的名字 --> <constructor-arg index="0" value="orderTopic"/> </bean> <!-- 消息监听,这里可以认为是A服务器的监听 --> <bean id="messageListener" class="com.raycloud.excalibur.mq.ConsumerMessageListener"/> <bean id="listenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="topicCustomerA" /> <property name="messageListener" ref="messageListener" /> <!-- 持久化消息 --> <property name="subscriptionDurable" value="true"/> <!-- 接收者ID --> <property name="clientId" value="clientA" /> <!-- 这里名字可以任意改变,A 领取了,你可以改成B 还可以领取,可以举例不是很恰当 --> <property name="durableSubscriptionName" value="clientA"/> </bean> </beans>
// 这是消费者代码,这里你可以创建 多个XMl文件,模拟多个消费者。 public class JmsTopicReceiver{ public static void main(String[] args) throws Exception { // 加载消费者监听 ApplicationContext context = new ClassPathXmlApplicationContext("spring/spring-jms-consumerA.xml"); // 写个死循环,模拟服务器一直运行 while (true){} } }
// 监听代码 直接输出 public class ConsumerMessageListener implements MessageListener { @Override public void onMessage(Message message) { System.out.println("topic 收到消息:"+message); } }
三、测试:
我们要完成一个收到一个order,然后N个服务计算的问题,因此采用topic 模式,而防止中途服务器挂掉,采用持久化方式,模拟测试如下:
1.启动两个ConsumerA,ConsumerB 监听,发布一个order ,同时收到消息,OK
2.启动一个ConsumerA,发布一个Order,再启动ConsumerB,也收到消息,OK
3.启动一个ConsumerA,发布一个Order,A收到,关闭mq服务器,重启mq服务(前后),重启ConsumerB ,同样收到消息,OK。
4.启动两个ConsumerA,ConsumerB,发布order,A,B收到消息,重启A,B 不收重复消息,OK
好像基本能满足需求了,由于发送量 不会很大,频率不会很高,可以试用一下了。
那么新问题是:
1.如果A,B 收到消息后,topic 的消息怎么处理呢? 一直保存着吗? 如果可以清除,怎么清除,什么时候进行清除呢?文件的方式方便管理吗?
2.虽然消息发送过去了,对象信息怎么接受呢,当然会有消息转换器..
3.在queue 模式下,服务器断开了,怎么从新连接呢,如果服务器挂了,怎么切换到备用的呢?
小结:
1.这是初步尝试了下 topic 持久化到文件,当然也可以持久化到数据的,关于activemq 持久化以及介绍文章,可以参考:http://blog.csdn.net/xyw_blog/article/details/9128219 比较详细。
2.
相关推荐
spring +activemq topic消息持久化订阅实例,整个项目中有activemq和spring的整合的所有实例,topic的持久化配置是在ApplicationContext3C、ApplicationContext3C2以及ApplicationContext3P三个中,消息生产者:...
spring+activemq必备jar包:activeio-core-3.1.4.jar,activemq-all-5.13.2.jar,activemq-pool-5.13.2.jar,commons-pool2-2.4.2.jar
spring boot activemq集成示例,包含queue和topic消息的发送、接收,连接池的支持。
ActiveMQ-demo实例代码: /* *生产者启动程序,并发送消息给amq服务器(broker) */ public class Producer { /** * @param args * @throws Exception */ public static void main(String[] args) throws ...
Windowss 下 Spring + ActiveMq + maven 集成配置分为两章,环境搭建之前参考了网上的博客以及同事的配置。
这是我从我工作项目中精简下来的Spring+ActiveMQ建议项目,代码亲测可行,也简单明了,很容易就能嵌入别的项目里面,已经通过这个demo嵌了2个实际项目。直接跑程序就可以发送并接受消息。前提要自己起MQ服务,至于MQ...
标签:activemq-spring-5.9.1-javadoc.jar,activemq,spring,5.9.1,javadoc,jar包下载,依赖包
spring+activemq,示例工程,演示JMS消息通信。。。。。。。
spring-boot 集成activemq代码demo
spring-activeMQ-演示 spring-activeMQ-演示 怎么跑? 1片spring-activeMQ-demo ``mvn全新安装`'' 2以tomcat7身份运行 3访问
spring activeMQ demo 配置
标签:activemq-spring-5.5.1-sources.jar,activemq,spring,5.5.1,sources,jar包下载,依赖包
Spring+ActiveMQ整合实例代码工程,朋友提供给我参考的,我备个份以便下次查阅,也分享给大家,看看对大家有没有帮助了
activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载
标签:activemq-spring-5.4.2-sources.jar,activemq,spring,5.4.2,sources,jar包下载,依赖包
使用spring jmstemplate写的activemq小demo,浅显易懂。工程下载导入可用(maven 工程) activemq 可直接apache官网下载 传送门http://activemq.apache.org/download.html
基于Spring+JMS+ActiveMQ+Tomcat,我使用的版本情况如下所示:Spring 3.2.0,ActiveMQ 5.4.3,Tomcat 6.0.43。本例通过详细的说明和注释,实现消息服务的基本功能:发送与接收。Spring对JMS提供了很好的支持,可以...
spring+activeMQ 嵌入式配置 完整demo(包含jar包)
Spring-ActiveMQ-jar
标签:activemq-web-demo-5.3.0-sources.jar,activemq,web,demo,5.3.0,sources,jar包下载,依赖包