ActiveMQ之釋出- 訂閱訊息模式實現
阿新 • • 發佈：2018-12-05
一、概念
- 釋出者/訂閱者模型支援向一個特定的訊息主題釋出訊息。0或多個訂閱者可能對接收來自特定訊息主題的訊息感興趣。在這種模型下，釋出者和訂閱者彼此不知道對方。這種模式好比是匿名公告板。這種模式被概括為：多個消費者可以獲得訊息
- 在釋出者和訂閱者之間存在時間依賴性。釋出者需要建立一個訂閱（subscription），以便客戶能夠訂閱。訂閱者必須保持持續的活動狀態以接收訊息，除非訂閱者建立了持久的訂閱。在那種情況下，在訂閱者未連線時釋出的訊息將在訂閱者重新連線時重新發布。
二、案例
2.1 訊息生產者-訊息釋出者
package com.shyroke.firstActiveMQ2; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 訊息生產者 * @author Administrator * */ public class JMSProducer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 預設的連線使用者名稱 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 預設的連線密碼 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 預設的連線地址 private static final int SENDNUM=10; // 傳送的訊息數量 public static void main(String[] args) { ConnectionFactory connectionFactory; // 連線工廠 Connection connection = null; // 連線 Session session; // 會話 接受或者傳送訊息的執行緒 Destination destination; // 訊息的目的地 MessageProducer messageProducer; // 訊息生產者 // 例項化連線工廠 connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通過連線工廠獲取連線 connection.start(); // 啟動連線 session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 建立Session // destination=session.createQueue("FirstQueue1"); // 建立訊息佇列 destination=session.createTopic("firstTopic"); messageProducer=session.createProducer(destination); // 建立訊息生產者 sendMessage(session, messageProducer); // 傳送訊息 session.commit(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } finally{ if(connection!=null){ try { connection.close(); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } } /** * 傳送訊息 * @param session * @param messageProducer * @throws Exception */ public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{ for(int i=0;i<JMSProducer.SENDNUM;i++){ TextMessage message=session.createTextMessage("ActiveMQ 釋出的訊息"+i); System.out.println("傳送訊息："+"ActiveMQ 釋出的訊息"+i); messageProducer.send(message); } } }
2.2 訊息消費者-訊息訂閱者一
package com.shyroke.firstActiveMQ2; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class JMSConsumer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 預設的連線使用者名稱 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 預設的連線密碼 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 預設的連線地址 public static void main(String[] args) { ConnectionFactory connectionFactory; // 連線工廠 Connection connection = null; // 連線 Session session; // 會話 接受或者傳送訊息的執行緒 Destination destination; // 訊息的目的地 MessageConsumer consumer; //建立消費者 // 例項化連線工廠 connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通過連線工廠獲取連線 connection.start(); // 啟動連線 /** * 這裡的最好使用Boolean.FALSE，如果是用true則必須commit才能生效，且http://127.0.0.1:8161/admin管理頁面才會更新訊息佇列的變化情況。 */ session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 建立Session // destination=session.createQueue("FirstQueue1"); // 建立訊息佇列 destination=session.createTopic("firstTopic"); consumer=session.createConsumer(destination); consumer.setMessageListener(new MyListener()); // 註冊訊息監聽 } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
-
監聽器1
package com.shyroke.firstActiveMQ2; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; public class MyListener implements MessageListener { public void onMessage(Message message) { try { System.out.println("訂閱者一收到的訊息："+((TextMessage)message).getText()); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
2.3 訊息消費者-訊息訂閱者二
package com.shyroke.firstActiveMQ2;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSConsumer2 {
private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 預設的連線使用者名稱
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 預設的連線密碼
private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 預設的連線地址
public static void main(String[] args) {
ConnectionFactory connectionFactory; // 連線工廠
Connection connection = null; // 連線
Session session; // 會話 接受或者傳送訊息的執行緒
Destination destination; // 訊息的目的地
MessageConsumer consumer; //建立消費者
// 例項化連線工廠
connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
try {
connection=connectionFactory.createConnection(); // 通過連線工廠獲取連線
connection.start(); // 啟動連線
/**
* 這裡的最好使用Boolean.FALSE，如果是用true則必須commit才能生效，且http://127.0.0.1:8161/admin管理頁面才會更新訊息佇列的變化情況。
*/
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 建立Session
// destination=session.createQueue("FirstQueue1"); // 建立訊息佇列
destination=session.createTopic("firstTopic");
consumer=session.createConsumer(destination);
consumer.setMessageListener(new MyListener2()); // 註冊訊息監聽
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
-
監聽器2
package com.shyroke.firstActiveMQ2;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class MyListener implements MessageListener {
public void onMessage(Message message) {
try {
System.out.println("訂閱者一收到的訊息："+((TextMessage)message).getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
-
測試以及結果
釋出- 訂閱訊息模式首先必須訂閱者先訂閱服務，然後釋出者再發布訊息，最後訂閱者受到服務訊息。
所以本章我們先執行訊息訂閱者一和訊息訂閱者二的程式碼然後檢視ActiveMQ的管理介面：
如圖可見，訂閱者已經註冊成功，然後再發布者釋出訊息：
