JMS学习七（ActiveMQ之Topic的持久订阅）

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

生产者：

package cn.slimsmart.activemq.demo.topic;  

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("tcp://192.168.18.43: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 cn.slimsmart.activemq.demo.topic;  

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;  

/**

 * 持久订阅设置唯一的客户端ID和订阅者ID。

 */

public class ConsumerPersistent {  

    public static void main(String[] args) throws JMSException {

        String clientId = "client_id";  

        // 连接到ActiveMQ服务器

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.18.43: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();

                }

            }

        });  

    }

}

注：

1.activemq区分消费者，是通过clientID和订阅客户名称来区分的。

2.消息的生产者，发送消息时用使用持久模式。
3.使用相同的“clientID”，则认为是同一个消费者。两个程序使用相同的“clientID”，则同时只能有一个连接到activemq，第二个连接的会报错。
4.activemq的设置在conf/activemq.xml中，默认消息是保存在data/kahadb中，重启activemq消息不会丢。

巴特西

JMS学习七（ActiveMQ之Topic的持久订阅）

最新文章

热门文章