非持久化訂閱持續(xù)到它們訂閱對象的生命周期。這意味著，客戶端只能在訂閱者活動時看到相關主題發(fā)布的消息。如果訂閱者不活動，它會錯過相關主題的消息。如果花費較大的開銷，訂閱者可以被定義為durable（持久化的）。持久化的訂閱者注冊一個帶有JMS保持的唯一標識的持久化訂閱（subscription）。帶有相同標識的后續(xù)訂閱者會再續(xù)前一個訂閱者的訂閱狀態(tài)。如果持久化訂閱沒有活動的訂閱者，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); // 創(chuàng)建主題 Topic topic = session.createTopic("slimsmart.topic.test"); MessageProducer producer = session.createProducer(topic); // NON_PERSISTENT 非持久化 PERSISTENT 持久化,發(fā)送消息時用使用持久模式 producer.setDeliveryMode(DeliveryMode.PERSISTENT); TextMessage message = session.createTextMessage(); message.setText("topic 消息。"); message.setStringProperty("property", "消息Property"); // 發(fā)布主題消息 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); // 創(chuàng)建主題 Topic topic = session.createTopic("slimsmart.topic.test"); // 創(chuàng)建持久訂閱,指定客戶端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區(qū)分消費者，是通過clientID和訂閱客戶名稱來區(qū)分的。

2.消息的生產者，發(fā)送消息時用使用持久模式。
3.使用相同的“clientID”，則認為是同一個消費者。兩個程序使用相同的“clientID”，則同時只能有一個連接到activemq，第二個連接的會報錯。
4.activemq的設置在conf/activemq.xml中，默認消息是保存在data/kahadb中，重啟activemq消息不會丟。

轉載于:https://www.cnblogs.com/alter888/p/8976221.html

總結

以上是生活随笔為你收集整理的JMS学习七（ActiveMQ之Topic的持久订阅）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。