ActiveMQ订阅模式持久化实现

实现步骤：
1、配置发送xml，applicationContext-send.xml

[html] view plain copy

1. <?xml version="1.0" encoding="UTF-8"?>
2. <beans xmlns="http://www.springframework.org/schema/beans"
3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
4. xsi:schemaLocation="http://www.springframework.org/schema/beans
5. http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
6. http://www.springframework.org/schema/context
7. http://www.springframework.org/schema/context/spring-context-2.5.xsd">
8. <context:property-placeholder location="classpath:/properties/jms.properties" />
9. <!-- 配置JMS连接工厂 -->
10. <bean id="myConnectionFactory"
11. class="org.springframework.jms.connection.CachingConnectionFactory">
12. <!-- Session缓存数量 -->
13. <property name="sessionCacheSize" value="10" />
14. <property name="targetConnectionFactory">
15. <bean class="org.apache.activemq.ActiveMQConnectionFactory">
16. <!-- MQ地址 -->
17. <property name="brokerURL" value="${brokerUrl}" />
18. <!-- 是否异步发送 -->
19. <property name="useAsyncSend" value="true" />
20. </bean>
21. </property>
22. </bean>
23. <!-- 发送消息的目的地（一个主题） -->
24. <bean id="myDestination" class="org.apache.activemq.command.ActiveMQTopic">
25. <!-- 设置消息主题的名字 -->
26. <constructor-arg index="0" value="${send.name}" />
27. </bean>
28. <!-- 配置JMS模版 -->
29. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
30. <property name="connectionFactory" ref="myConnectionFactory" />
31. <property name="defaultDestination" ref="myDestination" />
32. <!-- 订阅发布模式 -->
33. <property name="pubSubDomain" value="true" />
34. <property name="receiveTimeout" value="10000" />
35. </bean>
36. </beans>

2、编写发送java，ActiveMQsender.java

[java] view plain copy

1. package com.by.activeMQ;
2. import javax.jms.JMSException;
3. import javax.jms.Message;
4. import javax.jms.Session;
5. import javax.jms.TextMessage;
6. import org.springframework.context.ApplicationContext;
7. import org.springframework.context.support.ClassPathXmlApplicationContext;
8. import org.springframework.jms.core.JmsTemplate;
9. import org.springframework.jms.core.MessageCreator;
10. public class ActiveMQsender {
11. public static void main(String[] args) {
12. @SuppressWarnings("resource")
13. ApplicationContext ctx = new ClassPathXmlApplicationContext(
14. "ApplicationContext/applicationContext-send.xml");
15. JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
16. jmsTemplate.send(new MessageCreator() {
17. public Message createMessage(Session session) throws JMSException {
18. TextMessage msg = session.createTextMessage();
19. // 设置消息属性
20. msg.setStringProperty("mood", "happy");
21. // 设置消息内容
22. msg.setText("Hello World!");
23. return msg;
24. }
25. });
26. System.out.println("send end");
27. }
28. }

3、配置接收xml，applicationContext-receive.xml

[html] view plain copy

1. <?xml version="1.0" encoding="UTF-8"?>
2. <beans xmlns="http://www.springframework.org/schema/beans"
3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
4. xsi:schemaLocation="http://www.springframework.org/schema/beans
5. http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
6. http://www.springframework.org/schema/context
7. http://www.springframework.org/schema/context/spring-context-2.5.xsd">
8. <context:property-placeholder location="classpath:/properties/jms.properties" />
9. <!-- 第一个接收者 -->
10. <!-- 配置JMS连接工厂 -->
11. <bean id="myConnectionFactory"
12. class="org.springframework.jms.connection.CachingConnectionFactory">
13. <!-- Session缓存数量 -->
14. <property name="sessionCacheSize" value="10" />
15. <!-- 接收者ID -->
16. <property name="clientId" value="${topic.clientId}" />
17. <property name="targetConnectionFactory">
18. <bean class="org.apache.activemq.ActiveMQConnectionFactory">
19. <!-- MQ地址 -->
20. <property name="brokerURL" value="${brokerUrl}" />
21. </bean>
22. </property>
23. </bean>
24. <!-- 发送消息的目的地（一个主题） -->
25. <bean id="myDestination" class="org.apache.activemq.command.ActiveMQTopic">
26. <!-- 设置消息主题的名字 -->
27. <constructor-arg index="0" value="${topic.name}" />
28. </bean>
29. <!-- 生产消息配置 （自己定义）-->
30. <bean id="myTopicConsumer" class="com.by.activeMQ.ActiveMQreceiver" />
31. <!-- 消息监听器 -->
32. <bean id="myTopicListener"
33. class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
34. <constructor-arg ref="myTopicConsumer" />
35. <!-- 接收消息的方法名称 -->
36. <property name="defaultListenerMethod" value="receive" />
37. <!-- 不进行消息转换 -->
38. <property name="messageConverter"><null/></property>
39. </bean>
40. <!-- 消息监听容器 -->
41. <bean id="myListenerContainer"
42. class="org.springframework.jms.listener.DefaultMessageListenerContainer">
43. <property name="connectionFactory" ref="myConnectionFactory" />
44. <!-- 发布订阅模式 -->
45. <property name="pubSubDomain" value="true"/>
46. <!-- 消息持久化 -->
47. <property name="subscriptionDurable" value="true"/>
48. <property name="receiveTimeout" value="10"/>
49. <!-- 接收者ID -->
50. <property name="clientId" value="${topic.clientId}" />
51. <property name="durableSubscriptionName" value="${topic.clientId}"/>
52. <property name="destination" ref="myDestination" />
53. <property name="messageListener" ref="myTopicListener" />
54. </bean>
55. <!-- 第二个接收者 -->
56. <!-- 配置JMS连接工厂 -->
57. <bean id="myConnectionFactory2"
58. class="org.springframework.jms.connection.CachingConnectionFactory">
59. <!-- Session缓存数量 -->
60. <property name="sessionCacheSize" value="10" />
61. <!-- 接收者ID -->
62. <property name="clientId" value="${topic2.clientId}" />
63. <property name="targetConnectionFactory">
64. <bean class="org.apache.activemq.ActiveMQConnectionFactory">
65. <!-- MQ地址 -->
66. <property name="brokerURL" value="${brokerUrl}" />
67. </bean>
68. </property>
69. </bean>
70. <!-- 发送消息的目的地（一个主题） -->
71. <bean id="myDestination2" class="org.apache.activemq.command.ActiveMQTopic">
72. <!-- 设置消息主题的名字 -->
73. <constructor-arg index="0" value="${topic2.name}" />
74. </bean>
75. <!-- 生产消息配置 （自己定义）-->
76. <bean id="myTopicConsumer2" class="com.by.activeMQ.ActiveMQreceiver2" />
77. <!-- 消息监听器 -->
78. <bean id="myTopicListener2"
79. class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
80. <constructor-arg ref="myTopicConsumer2" />
81. <!-- 接收消息的方法名称 -->
82. <property name="defaultListenerMethod" value="receive" />
83. <!-- 不进行消息转换 -->
84. <property name="messageConverter"><null/></property>
85. </bean>
86. <!-- 消息监听容器 -->
87. <bean id="myListenerContainer2"
88. class="org.springframework.jms.listener.DefaultMessageListenerContainer">
89. <property name="connectionFactory" ref="myConnectionFactory2" />
90. <!-- 发布订阅模式 -->
91. <property name="pubSubDomain" value="true"/>
92. <!-- 消息持久化 -->
93. <property name="subscriptionDurable" value="true"/>
94. <property name="receiveTimeout" value="10"/>
95. <!-- 接收者ID -->
96. <property name="clientId" value="${topic2.clientId}" />
97. <property name="durableSubscriptionName" value="${topic2.clientId}"/>
98. <property name="destination" ref="myDestination2" />
99. <property name="messageListener" ref="myTopicListener2" />
100. </bean>
101. </beans>

4、编写接收java，ActiveMQreceiver.java

[java] view plain copy

1. package com.by.activeMQ;
2. import javax.jms.JMSException;
3. import javax.jms.TextMessage;
4. import org.springframework.jms.JmsException;
5. public class ActiveMQreceiver {
6. public void receive(TextMessage message) throws JmsException, JMSException {
7. String info = "this is receiver, "
8. + " mood is " + message.getStringProperty("mood") + ","
9. + "say " + message.getText();
10. System.out.println(info);
11. }
12. }

5、编写另一个接收java，ActiveMQreceiver.java

[java] view plain copy

1. package com.by.activeMQ;
2. import javax.jms.JMSException;
3. import javax.jms.TextMessage;
4. import org.springframework.jms.JmsException;
5. public class ActiveMQreceiver2 {
6. public void receive(TextMessage message) throws JmsException, JMSException {
7. String info = "this is receiver2,"
8. + " mood is " + message.getStringProperty("mood") + ","
9. + "say " + message.getText();
10. System.out.println(info);
11. }
12. }

6、编写一个main，开启接收监听，openReceive.java

[java] view plain copy

1. package com.by.activeMQ;
2. import org.springframework.context.ApplicationContext;
3. import org.springframework.context.support.ClassPathXmlApplicationContext;
4. public class openReceive {
5. public static void main(String[] args) {
6. @SuppressWarnings({ "unused", "resource" })
7. ApplicationContext ctx = new ClassPathXmlApplicationContext("ApplicationContext/applicationContext-receive.xml");
8. while(true) {
9. }
10. }
11. }

7、编写一个配置文件，jms.properties

[plain] view plain copy

1. #send
2. send.name=Topic_Mood
3. #receive
4. topic.name=Topic_Mood
5. topic.clientId=client_LiLei
6. topic2.name=Topic_Mood
7. topic2.clientId=client_HanMei
8. #url
9. brokerUrl=failover:(tcp://10.0.0.232:61616)?initialReconnectDelay=1000

8、pom里面添加activeMQ的依赖

[html] view plain copy

1. <dependency>
2. <groupId>org.apache.activemq</groupId>
3. <artifactId>activemq-pool</artifactId>
4. <version>5.11.1</version>
5. </dependency>
6. <dependency>
7. <groupId>org.apache.commons</groupId>
8. <artifactId>commons-pool2</artifactId>
9. <version>2.3</version>
10. </dependency>
11. <dependency>
12. <groupId>org.springframework</groupId>
13. <artifactId>spring-jms</artifactId>
14. <version>4.0.0.RELEASE</version>
15. </dependency>
16. <dependency>
17. <groupId>org.apache.activemq</groupId>
18. <artifactId>activemq-all</artifactId>
19. <version>5.11.1</version>
20. </dependency>

耶，运行就ok了。
发送消息的输出是这样的：

[plain] view plain copy

1. 2016-08-05 11:27:19 [ main:0 ] - [ INFO ] Refreshing org[email protected]16011db4: startup date [Fri Aug 05 11:27:19 CST 2016]; root of context hierarchy
2. 2016-08-05 11:27:19 [ main:31 ] - [ INFO ] Loading XML bean definitions from class path resource [ApplicationContext/applicationContext-send.xml]
3. 2016-08-05 11:27:19 [ main:187 ] - [ INFO ] Loading properties file from class path resource [properties/jms.properties]
4. 2016-08-05 11:27:19 [ main:392 ] - [ INFO ] Established shared JMS Connection: ActiveMQConnection {id=ID:MDG42V9PSU28IKA-60542-1470367639797-1:1,clientId=null,started=false}
5. 2016-08-05 11:27:19 [ ActiveMQ Task-1:467 ] - [ INFO ] Successfully connected to tcp://10.0.0.232:61616
6. send end

接收消息的输出是这样的：

[plain] view plain copy

1. 2016-08-05 11:28:04 [ ActiveMQ Task-1:490 ] - [ INFO ] Successfully connected to tcp://10.0.0.232:61616
2. 2016-08-05 11:28:04 [ main:498 ] - [ INFO ] Established shared JMS Connection: ActiveMQConnection {id=ID:MDG42V9PSU28IKA-60544-1470367684739-1:1,clientId=client_LiLei,started=false}
3. 2016-08-05 11:28:04 [ ActiveMQ Task-1:504 ] - [ INFO ] Successfully connected to tcp://10.0.0.232:61616
4. 2016-08-05 11:28:04 [ main:509 ] - [ INFO ] Established shared JMS Connection: ActiveMQConnection {id=ID:MDG42V9PSU28IKA-60544-1470367684739-3:1,clientId=client_HanMei,started=false}
5. this is receiver2, mood is happy,say Hello World!
6. this is receiver,  mood is happy,say Hello World!

配置另一个接收者就是，把第一个接收者的配置复制，然后添加个2，再把接收类复制，添加个2，就搞定了。这种方式也适用于mongodb啊这种配置。在一个工程里面操作两个mongodb数据库。