RocketMQ實現順序訊息
阿新 • • 發佈：2018-10-31
大部分業務場景不需要訊息的產生和消費的順序相同(順序消費)。但在某些場景（比如 訂單的生成，付款 ，下單 這3個訊息必須按順序處理）。解決思路是：（1）在傳送端，要把統一業務ID 傳送到同一個MessageQueue. (2) 在消費端，同一MessageQueue 的訊息不要被併發處理了。
針對（1） ， 我們可以使用RocketMQ 提供的 MesssageQueueSelector 介面以及預設的3種實現（例如SelectMessageQueueByHash, 你也可以自己實現這個介面），控制訊息向哪個MessageQueue 傳送。程式碼如下：
public class Producer { public static void main(String[] args) throws MQClientException { DefaultMQProducer producer = new DefaultMQProducer("orderProducerGroup"); producer.setNamesrvAddr("10.137.38.61:9876"); producer.start(); int orderId = 1; // 同一個訂單號 for (int i = 0; i < 20; i++) { try { Message msg = new Message("TopicA", "TagA", "Key" + i ,("OrderId" + orderId + ", Msg body" + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult = producer.send(msg, new SelectMessageQueueByHash(), orderId); System.out.println(sendResult.toString()); } catch (Exception e) { System.out.println(e); } } producer.shutdown(); } }
針對（2） 每一個 RocketMQ 提供了 MessageListenerOrderly 介面 作為一個RegisterMessageListener。 這種訊息監聽保證每次消費ConsumerQueue時會對每個ConsumerQueue 加鎖，從而保證同一個ConsumerQueue 的訊息不會多個Consumer 併發消費，又同樣保證了不同ConsumerQueue 的訊息可以被併發消費。這樣消費一個ConsumerQueue裡的訊息就保證了順序性。程式碼如下：
public class Consumer { public static void main(String[] args) throws MQClientException { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderConsumerGroup"); consumer.setNamesrvAddr("10.137.38.61:9876"); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); consumer.subscribe("TopicA", "TagA"); consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> { MessageExt msg = msgs.get(0); System.out.println(msg.toString()); return ConsumeOrderlyStatus.SUCCESS; }); consumer.start(); System.out.println("Consumer start."); } }
如下圖， consumer 端的 日誌， 顯示 所用的訊息orderId=1 的訊息都發送到了queueId=1 的 MessageQueue 中了。
