rocketmq有序訊息的（四）
阿新 • • 發佈：2021-11-11
opic的有序訊息已經成為mq的標配。而RocketMQ中是這樣區分訊息型別的， 普通訊息也叫做無序訊息，簡單來說就是沒有順序的訊息，而有序訊息就是按照一定的先後順序的訊息型別。舉個例子，producer 依次傳送 order id 為 1、2、3 的訊息到 broker，consumer 接到的訊息順序也就是 1、2、3 ，而不會出現普通訊息那樣的 2、1、3 等情況。
一、有序訊息該如何實現
理論上：我們都知道訊息首先由 producer 到 broker，再從 broker 到 consumer，分這兩步走。那麼要保證訊息的有序，勢必這兩步都是要保證有序的，即要保證訊息是按有序傳送到 broker，broker 也是有序將訊息投遞給 consumer，兩個條件必須同時滿足，缺一不可。
1.1、全域性有序訊息
由於一個 topic 只有一個 queue ，即使我們有多個 producer 例項和 consumer 例項也很難提高訊息吞吐量。就好比過獨木橋，大家只能一個挨著一個過去，效率低下。
1.2、區域性有序訊息
常見做法就是將 order id 進行處理，將 order id 相同的訊息傳送到 topicB 的同一個 queue，假設我們 topicB 有 2 個 queue，那麼我們可以簡單的對 id 取餘，奇數的發往 queue0，偶數的發往 queue1，消費者按照 queue 去消費時，就能保證 queue0 裡面的訊息有序消費，queue1 裡面的訊息有序消費。
二、RocketMQ的topic的補充
opic 只是訊息的邏輯分類，內部實現其實是由 queue 組成。當 producer 把訊息傳送到某個 topic 時，預設是會訊息傳送到具體的 queue 上。由於一個 topic 可以有多個 queue，所以在效能比全域性有序高得多。假設 queue 數是 n，理論上效能就是全域性有序的 n 倍，當然 consumer 也要跟著增加才行。在實際情況中，這種區域性有序訊息是會比全域性有序訊息用的更多。
/** * 有序訊息 */ public class OrderedProducer { public static final String NAME_SERVER_ADDR = "192.168.32.128:9876";public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException, UnsupportedEncodingException { // 1：建立生產者物件，並指定組名 DefaultMQProducer producer = new DefaultMQProducer("GROUP_TEST"); // 2：指定NameServer地址 producer.setNamesrvAddr(NAME_SERVER_ADDR); // 3：啟動生產者 producer.start(); producer.setRetryTimesWhenSendAsyncFailed(0); // 設定非同步傳送失敗重試次數，預設為2 // 4：定義訊息佇列選擇器 MessageQueueSelector messageQueueSelector = new MessageQueueSelector() { /** * 訊息佇列選擇器，保證同一條業務資料的訊息在同一個佇列 * @param mqs topic中所有佇列的集合 * @param msg 傳送的訊息 * @param arg 此引數是本示例中producer.send的第三個引數 * @return */ @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { Integer id = (Integer) arg; // id == 1001 int index = id % mqs.size(); // 分割槽順序：同一個模值的訊息在同一個佇列中 return mqs.get(index); // 全域性順序：所有的訊息都在同一個佇列中 // return mqs.get(mqs.size() - 1); } }; String[] tags = new String[]{"TagA", "TagB", "TagC"}; List<Map> bizDatas = getBizDatas(); // 5：迴圈傳送訊息 for (int i = 0; i < bizDatas.size(); i++) { Map bizData = bizDatas.get(i); // keys：業務資料的ID，比如使用者ID、訂單編號等等 Message msg = new Message("TopicTest", tags[i % tags.length], "" + bizData.get("msgType"), bizData.toString().getBytes(RemotingHelper.DEFAULT_CHARSET)); // 傳送有序訊息 SendResult sendResult = producer.send(msg, messageQueueSelector, bizData.get("msgType")); System.out.printf("%s， body:%s%n", sendResult, bizData); } // 6：關閉生產者 producer.shutdown(); } public static List<Map> getBizDatas() { List<Map> orders = new ArrayList<Map>(); HashMap orderData = new HashMap(); orderData.put("msgType", 1001); orderData.put("userId", "張三"); orderData.put("desc", "存錢1000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 2001); orderData.put("userId", "張三"); orderData.put("desc", "取錢1000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 3001); orderData.put("userId", "張三"); orderData.put("desc", "存錢2000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 4001); orderData.put("userId", "張三"); orderData.put("desc", "存錢3000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 5001); orderData.put("userId", "張三"); orderData.put("desc", "存錢4000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 6001); orderData.put("userId", "張三"); orderData.put("desc", "取錢5000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 7001); orderData.put("userId", "張三"); orderData.put("desc", "取錢6000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 8001); orderData.put("userId", "張三"); orderData.put("desc", "取錢2000"); orders.add(orderData); orderData = new HashMap(); orderData.put("msgType", 9001); orderData.put("userId", "張三"); orderData.put("desc", "存錢9000"); orders.add(orderData); return orders; } }
/** * 順序訊息消費者 */ public class OrderedConsumer { public static final String NAME_SERVER_ADDR = "192.168.32.128:9876"; public static void main(String[] args) throws Exception { // 1. 建立消費者（Push）物件 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("GROUP_TEST"); // 2. 設定NameServer的地址，如果設定了環境變數NAMESRV_ADDR，可以省略此步 consumer.setNamesrvAddr(NAME_SERVER_ADDR); /** * 設定Consumer第一次啟動是從佇列頭部開始消費還是佇列尾部開始消費<br> * 如果非第一次啟動，那麼按照上次消費的位置繼續消費 * 這裡設定的是一個consumer的消費策略 * CONSUME_FROM_LAST_OFFSET 預設策略，從該佇列最尾開始消費，即跳過歷史訊息 * CONSUME_FROM_FIRST_OFFSET 從佇列最開始開始消費，即歷史訊息（還儲存在broker的）全部消費一遍 * CONSUME_FROM_TIMESTAMP 從某個時間點開始消費，和setConsumeTimestamp()配合使用，預設是半個小時以前 */ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); // 3. 訂閱對應的主題和Tag String[] tags = new String[]{"TagA", "TagB", "TagC"}; consumer.subscribe("TopicTest", "TagA || TagB || TagC"); // 4. 註冊訊息接收到Broker訊息後的處理介面 // 注1：普通訊息消費 [[ // consumer.registerMessageListener(new MessageListenerConcurrently() { // AtomicInteger count = new AtomicInteger(0); // // public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { // doBiz(list.get(0)); // return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // } // }); // ]] 注1：普通訊息消費 // consumer consumer.setMaxReconsumeTimes(-1); // 延時 level 3 // 注2：順序訊息消費 [[ consumer.registerMessageListener(new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { context.setAutoCommit(true); doBiz(msgs.get(0)); return ConsumeOrderlyStatus.SUCCESS; } }); // 5. 啟動消費者(必須在註冊完訊息監聽器後啟動，否則會報錯) consumer.start(); System.out.println("已啟動消費者"); } /** * 模擬處理業務 * * @param message */ public static void doBiz(Message message) { try { System.out.printf("執行緒：%-25s 接收到新訊息 %s --- %s %n", Thread.currentThread().getName(), message.getTags(), new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET)); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } }這短短的一生我們最終都會失去，不妨大膽一點，愛一個人，攀一座山，追一個夢