ActiveMQ中topic类型消息的持久化简单测试

非持久化订阅持续到它们订阅对象的生命周期。这意味着，客户端只能在订阅者活动时看到相关主题发布的消息。如果订阅者不活动，它会错过相关主题的消息。如果花费较大的开销，订阅者可以被定义为durable（持久化的）。持久化的订阅者注册一个带有JMS保持的唯一标识的持久化订阅（subscription）。带有相同标识的后续订阅者会再续前一个订阅者的订阅状态。如果持久化订阅没有活动的订阅者，JMS会保持订阅消息，直到消息被订阅接收或者过期。

代码如下：

生产者：

package com.zzstxx.activemq.sample;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Producer {

	public static void main(String[] args) throws JMSException {
		// 连接到ActiveMQ服务器
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("username",
				"password","tcp://127.0.0.1:61616");  
		Connection connection = factory.createConnection();
		connection.start();
		Session session = connection.createSession(Boolean.TRUE,
				Session.AUTO_ACKNOWLEDGE);
		// 创建主题
		Topic topic = session.createTopic("slimsmart.topic.test");
		MessageProducer producer = session.createProducer(topic);
		// NON_PERSISTENT 非持久化 PERSISTENT 持久化,发送消息时用使用持久模式
		producer.setDeliveryMode(DeliveryMode.PERSISTENT);
		TextMessage message = session.createTextMessage();
		message.setText("topic 消息。");
		message.setStringProperty("property", "消息Property");
		// 发布主题消息
		producer.send(message);
		System.out.println("Sent message: " + message.getText());
		session.commit();
		session.close();
		connection.close();
	}
}

 

消费者：

package com.zzstxx.activemq.sample;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerPersistent {

	public static void main(String[] args) throws JMSException {  
        String clientId = "client_id";  
          
        // 连接到ActiveMQ服务器  
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("username",
				"password","tcp://127.0.0.1:61616");  
        Connection connection = factory.createConnection();  
        //客户端ID,持久订阅需要设置  
        connection.setClientID(clientId);  
        connection.start();  
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  
        // 创建主题  
        Topic topic = session.createTopic("slimsmart.topic.test");  
        // 创建持久订阅,指定客户端ID。  
        MessageConsumer consumer = session.createDurableSubscriber(topic,clientId);  
        consumer.setMessageListener(new MessageListener() {  
            // 订阅接收方法  
            public void onMessage(Message message) {  
                TextMessage tm = (TextMessage) message;  
                try {  
                    System.out.println("Received message: " + tm.getText()+":"+tm.getStringProperty("property"));  
                } catch (JMSException e) {  
                    e.printStackTrace();  
                }  
            }  
        });  
    }  
}