概要
- ActiveMQ 的 request-response通信方式
- ActiveMQ 的事務(wù)
- ActiveMQ消息消費的同步和異步
- ActiveMQ的零散知識點
ActiveMQ 的 request-response通信方式
在前面的兩種模式中都是一方負責(zé)發(fā)送消息而另外一方負責(zé)處理瑟匆。而我們實際中的很多應(yīng)用相當(dāng)于一種一應(yīng)一答的過程所禀，需要雙方都能給對方發(fā)送消息酌心。于是請求-應(yīng)答的這種通信方式也很重要检吆。它也應(yīng)用的很普遍。
首先發(fā)送端代碼
public class ReSend {
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination adminQueue = session.createQueue("shli.1");
//創(chuàng)建生產(chǎn)者發(fā)送消息
MessageProducer producer = session.createProducer(adminQueue);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//創(chuàng)建一個臨時的通道蛮瞄，并且創(chuàng)建消費者
Destination tempDest = session.createTemporaryQueue();
MessageConsumer responseConsumer = session.createConsumer(tempDest);
//設(shè)置臨時消息的一個監(jiān)聽器
responseConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
try {
System.out.println("收到回復(fù)內(nèi)容："+((TextMessage)arg0).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//模擬消息
TextMessage txtMessage = session.createTextMessage();
txtMessage.setText("我是ReSend");
//設(shè)置消息頭部信息 使得回復(fù)的消息可以定位到剛剛創(chuàng)建的臨時通道
txtMessage.setJMSReplyTo(tempDest);
//這里在消息頭部設(shè)置一個標(biāo)示嚼摩，好像僅僅是用于業(yè)務(wù)上的判斷，并不影響消息的應(yīng)答
String correlationId = UUID.randomUUID().toString();
txtMessage.setJMSCorrelationID(correlationId);
producer.send(txtMessage);
}
}
跟之前P2P不一樣的是發(fā)送端這里有創(chuàng)建一個臨時通道以及臨時通道對應(yīng)的消費者鲫剿，并且在消息的頭部添加了JMSReplyTo和MSCorrelationID，用于接收端進行消息的回復(fù)稻轨。
然后接收端的代碼
public class ReReceive {
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination adminQueue = session.createQueue("shli.1");
//創(chuàng)建回復(fù)用的生產(chǎn)者
final MessageProducer replyProducer = session.createProducer(null);
replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//創(chuàng)建消費者對消息進行消費
MessageConsumer consumer = session.createConsumer(adminQueue);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
TextMessage response = session.createTextMessage();
if (message instanceof TextMessage) {
TextMessage txtMsg = (TextMessage) message;
String messageText = txtMsg.getText();
System.out.println("收到了來自ReSend的消息------"+ messageText);
response.setText("回復(fù)ReSend灵莲，F(xiàn)rom ReReceive");
}
response.setJMSCorrelationID(message.getJMSCorrelationID());
replyProducer.send(message.getJMSReplyTo(), response);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
相比以之前的P2P接收端這里多了一個生產(chǎn)者
replyProducer，然后在收到消息之后殴俱，根據(jù)發(fā)送端發(fā)送過來的消息的頭部信息政冻，返回應(yīng)答信息~
ActiveMQ 的本地事務(wù)
Session，用于發(fā)送和接受消息线欲，而且是單線程的明场，支持事務(wù)的。比如之前的代碼
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
第一個參數(shù)為true的時候表示開啟事務(wù)李丰，為false的時候表示關(guān)閉事務(wù)苦锨。
在一個JMS客戶端，可以使用本地事務(wù)來組合消息的發(fā)送和接收趴泌。JMS Session接口提供了commit和rollback方法舟舒。事務(wù)提交意味著生產(chǎn)的所有消息被發(fā)送，消費的所有消息被確認(rèn)嗜憔；事務(wù)回滾意味著生產(chǎn)的所有消 息被銷毀秃励，消費的所有消息被恢復(fù)并重新提交，除非它們已經(jīng)過期吉捶。
事務(wù)性的會話總是牽涉到事務(wù)處理中夺鲜，commit或rollback方法一旦被調(diào)用，一個事務(wù)就結(jié)束了呐舔，而另一個事務(wù)被開始币励。關(guān)閉事務(wù)性會話將回滾其中的事務(wù)。
需要注意的是滋早，如果使用請求/回復(fù)機制榄审，即發(fā)送一個消息砌们，同時希望在同一個事務(wù)中等待接收該消息的回復(fù)杆麸，那么程序?qū)⒈粧炱鸶榻驗橹钡绞聞?wù)提交，發(fā)送操作才會真正執(zhí)行昔头。
需要注意的還有一個饼问，消息的生產(chǎn)和消費不能包含在同一個事務(wù)中。
createSession的第二個參數(shù)
在事務(wù)性會話中揭斧，當(dāng)一個事務(wù)被提交的時候莱革，確認(rèn)自動發(fā)生。在非事務(wù)性會話中讹开，消息何時被確認(rèn)取決于創(chuàng)建會話時的應(yīng)答模式（acknowledgement mode）盅视。該參數(shù)有以下三個可選值：
? Session.AUTO_ACKNOWLEDGE。當(dāng)客戶成功的從receive方法返回的時候旦万，或者從MessageListener.onMessage方法成功返回的時候闹击，會話自動確認(rèn)客戶收到的消息。
? Session.CLIENT_ACKNOWLEDGE成艘。 客戶通過消息的acknowledge方法確認(rèn)消息赏半。需要注意的是，在這種模式中淆两，確認(rèn)是在會話層上進行：確認(rèn)一個被消費的消息將自動確認(rèn)所有已被會話消 費的消息断箫。例如，如果一個消息消費者消費了10個消息秋冰，然后確認(rèn)第5個消息仲义，那么所有10個消息都被確認(rèn)。
JMS消息只有在被確認(rèn)之后丹莲，才認(rèn)為已經(jīng)被成功地消費了光坝。消息的成功消費通常包含三個階段：客戶接收消息、客戶處理消息和消息被確認(rèn)甥材。
? Session.DUPS_ACKNOWLEDGE盯另。 該選擇只是會話遲鈍第確認(rèn)消息的提交。如果JMS provider失敗洲赵，那么可能會導(dǎo)致一些重復(fù)的消息鸳惯。如果是重復(fù)的消息，那么JMS provider必須把消息頭的JMSRedelivered字段設(shè)置為true
ActiveMQ消息消費的同步和異步
同步接收消息：接收者主動接收消息叠萍，若消息隊列中沒有消息則阻塞等待芝发，當(dāng)然也有其它方法可以規(guī)定等待時間或是不等待。
Message message = messageConsumer1.receive();
這種方式線程會一直阻塞苛谷，直到接收到消息辅鲸，和Socket服務(wù)器端接收客戶端請求的方式差不多
異步接收消息：當(dāng)消息隊列有消息時會調(diào)用接收者的onMessage方法，接收者不用阻塞等待腹殿，可執(zhí)行其它業(yè)務(wù)
messageConsumer1.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
MapMessage map = (MapMessage)arg0;
try {
String shli = map.getString("name");
double price = map.getDouble("price");
boolean up = map.getBoolean("up");
System.out.println(shli + "-1-1-1-" + price +"-1--1-1--"+up+"---Thread---" +Thread.currentThread().getName());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
這種方式不會出現(xiàn)線程阻塞独悴，當(dāng)消息到達時例书，會在其他線程中執(zhí)行onMessage()中的邏輯
ActiveMQ的零散功能點
消息過期 和 優(yōu)先級
可以設(shè)置消息在一定時間后過期，默認(rèn)是永不過期刻炒。設(shè)置方法 后兩個send方法的最后一個參數(shù)
可以使用消息優(yōu)先級來指示JMS provider首先提交緊急的消息决采。優(yōu)先級分10個級別，從0（最低）到9（最高）坟奥。如果不指定優(yōu)先級树瞭，默認(rèn)級別是4。需要注意的是爱谁，JMS provider并不一定保證按照優(yōu)先級的順序提交消息晒喷。設(shè)置方法 后兩個send方法的最后第三個參數(shù)
JMS Selectors
消息對象有消息屬性，用于消息選擇器
使用如下：
-
生產(chǎn)者端需要設(shè)置消息屬性访敌，一定要注意的是setXxxProperty(filed,value)
message.setIntProperty("pa", 6);
-
給出條件厨埋，其實本質(zhì)上就是SQL92語法
String condition = "pa > 7";
-
創(chuàng)建消費者的時候，指定條件即可
MessageConsumer messageConsumer1 = session.createConsumer(destination,condition);
持久性
? PERSISTENT捐顷。指示JMS provider持久保存消息荡陷，以保證消息不會因為JMS provider的失敗而丟失。
? NON_PERSISTENT迅涮。不要求JMS provider持久保存消息
Producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
持久化訂閱
在第一篇的代碼中我們發(fā)現(xiàn)废赞，在訂閱模式中，如果消費者不在線叮姑，在上線之后是沒有辦法重新收到生產(chǎn)者發(fā)送的消息的唉地。那是因為我們沒有將消費者設(shè)置為持久化訂閱。
客戶可以通過會話上的createDurableSubscriber方法來創(chuàng)建一個持久訂閱传透，該方法的第一個參數(shù)必須是一個topic耘沼。第二個參數(shù)是訂閱的名稱。
這樣設(shè)置之后朱盐，消費者上線后就可以收到之前生產(chǎn)者發(fā)送的（離線）消息群嗤。
但是這種沒有持久性的消息只是被緩存在內(nèi)存中，如果ActiveMQ被關(guān)閉并重新打開的話兵琳，消費者同樣接收不到(離線)消息狂秘。
如果在發(fā)送端加上一句代碼，這樣就算是ActiveMQ重啟也不會出現(xiàn)消息丟失的現(xiàn)象了
Producer.setDeliveryMode(DeliveryMode.PERSISTENT);
JMS provider會存儲發(fā)布到持久訂閱對應(yīng)的topic上的消息躯肌。如果最初創(chuàng)建持久訂閱的客戶或者任何其它客戶使用相同的連接工廠和連接的客戶ID者春、相同 的主題和相同的訂閱名再次調(diào)用會話上的createDurableSubscriber方法，那么該持久訂閱就會被激活清女。JMS provider會象客戶發(fā)送客戶處于非激活狀態(tài)時所發(fā)布的消息钱烟。
持久訂閱在某個時刻只能有一個激活的訂閱者。持久訂閱在創(chuàng)建之后會一直保留，直到應(yīng)用程序調(diào)用會話上的unsubscribe方法拴袭。
最后
下一遍將記錄ActiveMQ和其他框架的整合
