ActiveMQ消息持久化
2017-11-21 00:15
513 查看
ActiveMQ消息持久化
消息持久化就是将消息保存到磁盘，这样的好处就是如果服务挂了，则消息还保存在磁盘不会丢失，服务起来后还能找到消息并在此发送，消息的持久化和消息的发送模型是没有关系的。
[html] view
plain copy
<!-- spring 使用jmsTemplate来实现消息的发送和接受 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="defaultDestination" ref="destination"></property>
<property name="messageConverter">
<bean
class="org.springframework.jms.support.converter.SimpleMessageConverter" />
</property>
<!--开启订阅模式-->
<property name="pubSubDomain" value="false"/>
<property name="sessionAcknowledgeMode" value="1" />
<!-- deliveryMode, priority, timeToLive 的开关，要生效，必须配置explicitQosEnabled为true，默认false-->
<property name="explicitQosEnabled" value="true" />
<!-- 发送模式 DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久-->
<property name="deliveryMode" value="2"/>
</bean>
消息持久化配置如上，具体持久化到数据库还是文件还是在消息服务的activemq.xml中配置！
ok 消息的持久化搞定了我们来看看Topic的主题持久化订阅的实现。
Topic消息持久化订阅
通过之前的学习我们知道，消息的持久化订阅的要求：（1）、消息持久化 （2）、消息消费端要指定ClientID，同一时刻只能有一个ClientID相同的消费者连接消费，所以如果有多个消费者，则ClientID不能相同。下面来看看！
项目结构如下：
项目结构介绍：
1、一个消息生产者，两个消息消费者和一个监听类即MyMessageListener.java，对消息的异步处理和重Activemq时编写的一样！
2、三个配置文件即消息生产者的配置文件和消息消费者的配置文件。
其他的就是之前的相关demo这里相关的就上面1、2中提到的相关东西！
其实东西确实很少，我们来看看代码:
1、消息生产者配置文件即ApplicationContext3P.xml配置文件：
[html] view
plain copy
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd">
<!--第三方工厂 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
<property name="useAsyncSend" value="true" />
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="100" />
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
</bean>
<!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic" />
</bean>
<!-- spring 使用jmsTemplate来实现消息的发送和接受 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="defaultDestination" ref="destinationTopic"></property>
<!-- 进行持久化 -->
<property name="deliveryMode" value="2" />
<!-- 开启订阅模式 -->
<property name="pubSubDomain" value="true" />
</bean>
</beans>
2、消息生产者：
[java] view
plain copy
package springs.activemq.Service;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
/**
* @author Administrator
*
*/
public class QueueProducer {
// 负责消息的发送和接收可以理解为MessageProducer 和MessageConsummer的组合。
private static JmsTemplate jt = null;
public static void main(String[] args) {
ApplicationContext ctx = new ClassPathXmlApplicationContext(
"config/ApplicationContext3P.xml");
// 获取JmsTemplate对象
jt = (JmsTemplate) ctx.getBean("jmsTemplate");
// 调用方法，发送消息
jt.send(new MessageCreator() {
// 消息的产生，返回消息发送消息
public Message createMessage(Session s) throws JMSException {
TextMessage msg = s
.createTextMessage("Spring send msg ----> Hello activeMQ3");
return msg;
}
});
System.out.println("end!");
}
}
3、消息消费者1的配置文件：
[html] view
plain copy
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd">
<!--第三方工厂 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
<property name="useAsyncSend" value="true" />
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="100" />
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<!--消费者标示id -->
<property name="clientId" value="clientId_001" />
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
</bean>
<!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic" />
</bean>
<!--消息消费者监听类 -->
<bean id="myMessageListener" class="springs.activemq.Service.MyMessageListener" />
<!--监听容器的配置 -->
<bean id="myListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<!--消息目的地 -->
<property name="destination" ref="destinationTopic" />
<!--消息监听类 -->
<property name="messageListener" ref="myMessageListener" />
<!-- 发布订阅模式 -->
<property name="pubSubDomain" value="true" />
<!-- 消息持久化值设置为true -->
<property name="subscriptionDurable" value="true" />
<!--消息接收超时 -->
<property name="receiveTimeout" value="10000" />
<!-- 接收者ID -->
<property name="clientId" value="clientId_001" />
<property name="durableSubscriptionName" value="clientId_001" />
</bean>
</beans>
4、消息消费者2的配置文件：
[html] view
plain copy
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd">
<!--第三方工厂 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
<property name="useAsyncSend" value="true" />
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="100" />
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<!--消费者标示id -->
<property name="clientId" value="clientId_002" />
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
</bean>
<!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic" />
</bean>
<!--消息消费者监听类 -->
<bean id="myMessageListener" class="springs.activemq.Service.MyMessageListener" />
<!--监听容器的配置 -->
<bean id="myListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<!--消息目的地 -->
<property name="destination" ref="destinationTopic" />
<!--消息监听类 -->
<property name="messageListener" ref="myMessageListener" />
<!-- 发布订阅模式 -->
<property name="pubSubDomain" value="true" />
<!-- 消息持久化值设置为true -->
<property name="subscriptionDurable" value="true" />
<!--消息接收超时 -->
<property name="receiveTimeout" value="10000" />
<!-- 接收者ID -->
<property name="clientId" value="clientId_002" />
<property name="durableSubscriptionName" value="clientId_002" />
</bean>
</beans>
5、消费者1即SimpleJMSReceiver.java
[java] view
plain copy
package springs.activemq.Service;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class SimpleJMSReceiver {
public static void main(String[] args) {
ApplicationContext ctx = new ClassPathXmlApplicationContext("config/ApplicationContext3C.xml");
while(true) {
}
}
}
6、消息消费者2SimpleJMSReceiver2.java
[java] view
plain copy
package springs.activemq.Service;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class SimpleJMSReceiver2 {
public static void main(String[] args) {
ApplicationContext ctx = new ClassPathXmlApplicationContext("config/ApplicationContext3C2.xml");
while(true) {
}
}
}
7、消息监听类
[java] view
plain copy
package springs.activemq.Service;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class MyMessageListener implements MessageListener {
public void onMessage(Message arg0) {
// TODO Auto-generated method stub
try {
String message = ((TextMessage) arg0).getText();
System.out.println("textmessage:" + message);
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
消息监听类的onMessage方法的参数就是接收到的消息，我们在方法中对消息进行业务处理！
ok 到这里消息的持久化订阅就ok了，简单说一下：
1、Topic的消息持久化订阅需要将消息持久化，可以到文件也可以到数据库，这个看消息服务器中activemq.xml中持久化的配置，而发送的消息是否要持久化需要在jmsTemplate中配置"deliveryMode"的值为2.
2、消息消费者要设置clientid标识符，这样才能知道是谁。
3、消息监听类，这个类是来处理消息的，在实际项目中使用同步的应该不多，应该都是这样的即配置监听类，比如这个demo我们的监听类用了同一个，但是在项目中如果有不同的需求我们可以对每个消费者定义各个监听类来做具体的处理！
ok到这里也就完了，可以启动消息生产者生产消息，持久化到数据库中，然后启动消费者1和消费者2来消费消息，你可以看到两个消费者都能获取到消息。这时再看数据库则数据库中的消息都没了！
执行结果：
1、消息生产者：
2、消息持久化到mysql数据库中：
3、启动消费者1：
4、启动消费者2
ok这样整个流程都走完了，当消息被发送到消息服务后，如果服务挂了，你在启动消息服务之后再启动消息消费者，这时还是能正常获取未消费的消息。这样也就满足我们的持久化订阅了！
消息持久化就是将消息保存到磁盘，这样的好处就是如果服务挂了，则消息还保存在磁盘不会丢失，服务起来后还能找到消息并在此发送，消息的持久化和消息的发送模型是没有关系的。
[html] view
plain copy
<!-- spring 使用jmsTemplate来实现消息的发送和接受 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="defaultDestination" ref="destination"></property>
<property name="messageConverter">
<bean
class="org.springframework.jms.support.converter.SimpleMessageConverter" />
</property>
<!--开启订阅模式-->
<property name="pubSubDomain" value="false"/>
<property name="sessionAcknowledgeMode" value="1" />
<!-- deliveryMode, priority, timeToLive 的开关，要生效，必须配置explicitQosEnabled为true，默认false-->
<property name="explicitQosEnabled" value="true" />
<!-- 发送模式 DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久-->
<property name="deliveryMode" value="2"/>
</bean>
消息持久化配置如上，具体持久化到数据库还是文件还是在消息服务的activemq.xml中配置！
ok 消息的持久化搞定了我们来看看Topic的主题持久化订阅的实现。
Topic消息持久化订阅
通过之前的学习我们知道，消息的持久化订阅的要求：（1）、消息持久化 （2）、消息消费端要指定ClientID，同一时刻只能有一个ClientID相同的消费者连接消费，所以如果有多个消费者，则ClientID不能相同。下面来看看！
项目结构如下：
项目结构介绍：
1、一个消息生产者，两个消息消费者和一个监听类即MyMessageListener.java，对消息的异步处理和重Activemq时编写的一样！
2、三个配置文件即消息生产者的配置文件和消息消费者的配置文件。
其他的就是之前的相关demo这里相关的就上面1、2中提到的相关东西！
其实东西确实很少，我们来看看代码:
1、消息生产者配置文件即ApplicationContext3P.xml配置文件：
[html] view
plain copy
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd">
<!--第三方工厂 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
<property name="useAsyncSend" value="true" />
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="100" />
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
</bean>
<!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic" />
</bean>
<!-- spring 使用jmsTemplate来实现消息的发送和接受 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="defaultDestination" ref="destinationTopic"></property>
<!-- 进行持久化 -->
<property name="deliveryMode" value="2" />
<!-- 开启订阅模式 -->
<property name="pubSubDomain" value="true" />
</bean>
</beans>
2、消息生产者：
[java] view
plain copy
package springs.activemq.Service;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
/**
* @author Administrator
*
*/
public class QueueProducer {
// 负责消息的发送和接收可以理解为MessageProducer 和MessageConsummer的组合。
private static JmsTemplate jt = null;
public static void main(String[] args) {
ApplicationContext ctx = new ClassPathXmlApplicationContext(
"config/ApplicationContext3P.xml");
// 获取JmsTemplate对象
jt = (JmsTemplate) ctx.getBean("jmsTemplate");
// 调用方法，发送消息
jt.send(new MessageCreator() {
// 消息的产生，返回消息发送消息
public Message createMessage(Session s) throws JMSException {
TextMessage msg = s
.createTextMessage("Spring send msg ----> Hello activeMQ3");
return msg;
}
});
System.out.println("end!");
}
}
3、消息消费者1的配置文件：
[html] view
plain copy
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd">
<!--第三方工厂 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
<property name="useAsyncSend" value="true" />
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="100" />
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<!--消费者标示id -->
<property name="clientId" value="clientId_001" />
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
</bean>
<!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic" />
</bean>
<!--消息消费者监听类 -->
<bean id="myMessageListener" class="springs.activemq.Service.MyMessageListener" />
<!--监听容器的配置 -->
<bean id="myListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<!--消息目的地 -->
<property name="destination" ref="destinationTopic" />
<!--消息监听类 -->
<property name="messageListener" ref="myMessageListener" />
<!-- 发布订阅模式 -->
<property name="pubSubDomain" value="true" />
<!-- 消息持久化值设置为true -->
<property name="subscriptionDurable" value="true" />
<!--消息接收超时 -->
<property name="receiveTimeout" value="10000" />
<!-- 接收者ID -->
<property name="clientId" value="clientId_001" />
<property name="durableSubscriptionName" value="clientId_001" />
</bean>
</beans>
4、消息消费者2的配置文件：
[html] view
plain copy
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd">
<!--第三方工厂 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
<property name="useAsyncSend" value="true" />
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="100" />
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<!--消费者标示id -->
<property name="clientId" value="clientId_002" />
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
</bean>
<!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic" />
</bean>
<!--消息消费者监听类 -->
<bean id="myMessageListener" class="springs.activemq.Service.MyMessageListener" />
<!--监听容器的配置 -->
<bean id="myListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<!--消息目的地 -->
<property name="destination" ref="destinationTopic" />
<!--消息监听类 -->
<property name="messageListener" ref="myMessageListener" />
<!-- 发布订阅模式 -->
<property name="pubSubDomain" value="true" />
<!-- 消息持久化值设置为true -->
<property name="subscriptionDurable" value="true" />
<!--消息接收超时 -->
<property name="receiveTimeout" value="10000" />
<!-- 接收者ID -->
<property name="clientId" value="clientId_002" />
<property name="durableSubscriptionName" value="clientId_002" />
</bean>
</beans>
5、消费者1即SimpleJMSReceiver.java
[java] view
plain copy
package springs.activemq.Service;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class SimpleJMSReceiver {
public static void main(String[] args) {
ApplicationContext ctx = new ClassPathXmlApplicationContext("config/ApplicationContext3C.xml");
while(true) {
}
}
}
6、消息消费者2SimpleJMSReceiver2.java
[java] view
plain copy
package springs.activemq.Service;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class SimpleJMSReceiver2 {
public static void main(String[] args) {
ApplicationContext ctx = new ClassPathXmlApplicationContext("config/ApplicationContext3C2.xml");
while(true) {
}
}
}
7、消息监听类
[java] view
plain copy
package springs.activemq.Service;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class MyMessageListener implements MessageListener {
public void onMessage(Message arg0) {
// TODO Auto-generated method stub
try {
String message = ((TextMessage) arg0).getText();
System.out.println("textmessage:" + message);
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
消息监听类的onMessage方法的参数就是接收到的消息，我们在方法中对消息进行业务处理！
ok 到这里消息的持久化订阅就ok了，简单说一下：
1、Topic的消息持久化订阅需要将消息持久化，可以到文件也可以到数据库，这个看消息服务器中activemq.xml中持久化的配置，而发送的消息是否要持久化需要在jmsTemplate中配置"deliveryMode"的值为2.
2、消息消费者要设置clientid标识符，这样才能知道是谁。
3、消息监听类，这个类是来处理消息的，在实际项目中使用同步的应该不多，应该都是这样的即配置监听类，比如这个demo我们的监听类用了同一个，但是在项目中如果有不同的需求我们可以对每个消费者定义各个监听类来做具体的处理！
ok到这里也就完了，可以启动消息生产者生产消息，持久化到数据库中，然后启动消费者1和消费者2来消费消息，你可以看到两个消费者都能获取到消息。这时再看数据库则数据库中的消息都没了！
执行结果：
1、消息生产者：
2、消息持久化到mysql数据库中：
3、启动消费者1：
4、启动消费者2
ok这样整个流程都走完了，当消息被发送到消息服务后，如果服务挂了，你在启动消息服务之后再启动消息消费者，这时还是能正常获取未消费的消息。这样也就满足我们的持久化订阅了！
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- ActiveMQ入门到精通4-消息持久化到Mysql
- JMS消息持久化，将ActiveMQ消息持久化到mySql数据库中
- JMS学习(七)-ActiveMQ消息的持久存储方式之KahaDB存储
- JMS之——ActiveMQ消息持久化
- ActiveMQ -- 消息持久化
- ActiveMQ - 持久化消息与持久主题订阅
- JMS学习(五)--ActiveMQ中的消息的持久化和非持久化 以及 持久订阅者 和 非持久订阅者之间的区别与联系
- ActiveMQ学习记录 之 消息持久化
- JMS消息持久化，将ActiveMQ消息持久化到mySql数据库中
- springboot整合activemq，应答模式，消息重发机制，消息持久化
- ActiveMQ使用MySQL进行消息持久化
- 3、ActiveMQ学习记录 之 消息持久化
- ActiveMQ的消息持久化到Mysql数据库
- MQTT下ActiveMQ的消息持久化
- JMS之activemq消息持久化
- ActiveMQ消息持久化到Mysql数据库
- 快速上手MQ之activeMQ(4)--消息持久化
- [jms那点事儿] JMS ( java message service )入门 + activemq消息持久化
- [jms那点事儿] JMS ( java message service )入门 + activemq消息持久化
- ActiveMQ开启用户认证及消息持久化到数据库功能