ActiveMQ（二）——訊息處理機制
阿新 • • 發佈：2018-12-18
一、前言 上文中，小編提到安裝ActiveMQ，但是對於ActiveMQ中訊息是用什麼樣的形式儲存的？下面小編就向大家介紹一下。
二、訊息型別 對於訊息的傳遞有兩種型別：
1.點對點的，即一個生產者和一個消費者一一對應；
2.釋出/訂閱模式，即一個生產者產生訊息並進行傳送後，可以由多個消費者進行接收。
JMS定義了五種不同的訊息正文格式，以及呼叫的訊息型別，允許你傳送並接收以一些不同形式的資料，提供現有訊息格式的一些級別的相容性。
StreamMessage – Java原始值的資料流 MapMessage–一套名稱-值對 TextMessage–一個字串物件 ObjectMessage–一個序列化的 Java物件 BytesMessage–一個位元組的資料流 三、點對點Queue
第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號。 第二步：使用ConnectionFactory物件建立一個Connection物件。 第三步：開啟連線，呼叫Connection物件的start方法。 第四步：使用Connection物件建立一個Session物件。 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。 第六步：使用Session物件建立一個Producer物件。 第七步：建立一個Message物件，建立一個TextMessage物件。 第八步：使用Producer物件傳送訊息。 第九步：關閉資源。
@Test public void testQueueProducer() throws Exception { // 第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號。 //brokerURL伺服器的ip及埠號 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616"); // 第二步：使用ConnectionFactory物件建立一個Connection物件。 Connection connection = connectionFactory.createConnection(); // 第三步：開啟連線，呼叫Connection物件的start方法。 connection.start(); // 第四步：使用Connection物件建立一個Session物件。 //第一個引數：是否開啟事務。true：開啟事務，第二個引數忽略。 //第二個引數：當第一個引數為false時，才有意義。訊息的應答模式。1、自動應答2、手動應答。一般是自動應答。 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。 //引數：佇列的名稱。 Queue queue = session.createQueue("test-queue"); // 第六步：使用Session物件建立一個Producer物件。 MessageProducer producer = session.createProducer(queue); // 第七步：建立一個Message物件，建立一個TextMessage物件。 /*TextMessage message = new ActiveMQTextMessage(); message.setText("hello activeMq,this is my first test.");*/ TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test."); // 第八步：使用Producer物件傳送訊息。 producer.send(textMessage); // 第九步：關閉資源。 producer.close(); session.close(); connection.close(); }
3.2 12.1.2. Consumer 消費者：接收訊息。
第一步：建立一個ConnectionFactory物件。
第二步：從ConnectionFactory物件中獲得一個Connection物件。
第三步：開啟連線。呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件。和傳送端保持一致queue，並且佇列的名稱一致。
第六步：使用Session物件建立一個Consumer物件。
第七步：接收訊息。
第八步：列印訊息。
第九步：關閉資源
@Test
public void testQueueConsumer() throws Exception {
// 第一步：建立一個ConnectionFactory物件。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：從ConnectionFactory物件中獲得一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線。呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件。和傳送端保持一致queue，並且佇列的名稱一致。
Queue queue = session.createQueue("test-queue");
// 第六步：使用Session物件建立一個Consumer物件。
MessageConsumer consumer = session.createConsumer(queue);
// 第七步：接收訊息。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
String text = null;
//取訊息的內容
text = textMessage.getText();
// 第八步：列印訊息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//等待鍵盤輸入
System.in.read();
// 第九步：關閉資源
consumer.close();
session.close();
connection.close();
}
四、釋出訂閱Topic 4.1 Producer 使用步驟：
第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號。
第二步：使用ConnectionFactory物件建立一個Connection物件。
第三步：開啟連線，呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Topic物件。
第六步：使用Session物件建立一個Producer物件。
第七步：建立一個Message物件，建立一個TextMessage物件。
第八步：使用Producer物件傳送訊息。
第九步：關閉資源。
@Test
public void testTopicProducer() throws Exception {
// 第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號。
// brokerURL伺服器的ip及埠號
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：使用ConnectionFactory物件建立一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線，呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
// 第一個引數：是否開啟事務。true：開啟事務，第二個引數忽略。
// 第二個引數：當第一個引數為false時，才有意義。訊息的應答模式。1、自動應答2、手動應答。一般是自動應答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個topic物件。
// 引數：話題的名稱。
Topic topic = session.createTopic("test-topic");
// 第六步：使用Session物件建立一個Producer物件。
MessageProducer producer = session.createProducer(topic);
// 第七步：建立一個Message物件，建立一個TextMessage物件。
/*
* TextMessage message = new ActiveMQTextMessage(); message.setText(
* "hello activeMq,this is my first test.");
*/
TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
// 第八步：使用Producer物件傳送訊息。
producer.send(textMessage);
// 第九步：關閉資源。
producer.close();
session.close();
connection.close();
}
4.2 Consumer 消費者：接收訊息。
第一步：建立一個ConnectionFactory物件。
第二步：從ConnectionFactory物件中獲得一個Connection物件。
第三步：開啟連線。呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件。和傳送端保持一致topic，並且話題的名稱一致。
第六步：使用Session物件建立一個Consumer物件。
第七步：接收訊息。
第八步：列印訊息。
第九步：關閉資源
@Test
public void testTopicConsumer() throws Exception {
// 第一步：建立一個ConnectionFactory物件。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：從ConnectionFactory物件中獲得一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線。呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件。和傳送端保持一致topic，並且話題的名稱一致。
Topic topic = session.createTopic("test-topic");
// 第六步：使用Session物件建立一個Consumer物件。
MessageConsumer consumer = session.createConsumer(topic);
// 第七步：接收訊息。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
String text = null;
// 取訊息的內容
text = textMessage.getText();
// 第八步：列印訊息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
System.out.println("topic的消費端03。。。。。");
// 等待鍵盤輸入
System.in.read();
// 第九步：關閉資源
consumer.close();
session.close();
connection.close();
}
五、小結 通過分析和用不與spring結合的測試方法來更加深刻的瞭解mq在具體的使用過程中的步驟。
都是引入相關jar包後，建立連線工廠，工廠建立相關的連線物件，連線物件開啟，建立相關的會話，會話發起目的地並且建立生產者或者消費者，然後對訊息進行處理，最後關閉連線。
下一篇部落格小編會向大家介紹如何使用spring管理ActiveMQ。