2. 程式人生 > >RocketMQ-順序訊息

RocketMQ-順序訊息

阿新 • • 發佈：2018-12-12

github程式碼下載地址

1.RocketMQ有三種訊息

    1)普通訊息

    2)順序訊息

    3)事務訊息

2.順序訊息

    概念：是MQ提供的一種嚴格按照順序進行釋出和消費的訊息型別；因此可以看出，順序訊息由兩部分組成，順序釋出和

順序消費。

3.在MQ中如何保證順序消費

        (1) 訊息傳送是保證是順序的    (2) 訊息被儲存時保證是和傳送的順序一致 (3)訊息被消費時保證和儲存的順序一致

傳送訊息保證是順序的意味著對於有順序的訊息，使用者要保證使用同一個執行緒採用同步的方式傳送；而儲存和傳送的

順序一致則要求在同一個執行緒傳送過來的訊息A和訊息B，儲存時在空間上訊息A一定在訊息B之前；最後消費和儲存的順

序保持一致則要求，在訊息A和訊息B到達Consumer之後必須按照先消費訊息A在消費訊息B的順序執行

對於兩個訂單的訊息的原始資料：a1、b1、b2、a2、a3、b3（絕對時間下發生的順序）:

   1) 在傳送時，a訂單的訊息必須保證a1 ,a2 , a3的順序傳送，b訂單的訊息也一樣，但是a，b訂單之間的訊息沒有順序關係；

這意味這a，b訂單的訊息可以通過不同的執行緒傳送出去

   2) 在儲存時，需要分別保證a，b訂單各自訊息的順序，但是a，b訂單之間的訊息可以不保證

   3)在消費時，只要保證每一個分割槽只有一個執行緒來去處理即可當然，如果a、b在一個分割槽中，在收到訊息後也可以將他們拆分到不同執行緒中處理，不過要權衡一下收益

4.生產者程式碼

      1)生產者

package com.roger.order.producer;

import com.roger.order.entity.OrderMsgDTO;
import com.roger.utils.SnowflakeIdWorker;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.ArrayList;
import java.util.List;

public class OrderMqProducer {

    public static void main(String[] args) throws Exception {
        DefaultMQProducer defaultMQProducer =
                new DefaultMQProducer("orderMQProducerGroup");
        defaultMQProducer.setNamesrvAddr("172.20.10.60:9876");
        defaultMQProducer.start();
        defaultMQProducer.createTopic("OrderTopic", "OrderTopic", 3);

        String[] tags = new String[]{"TagC", "TagP", "TagF"};

        List<OrderMsgDTO> orderMsgList = new ArrayList<>();
        int orderCount = 5;
        for (int i = 0; i < orderCount; i++) {
            long orderId = SnowflakeIdWorker.getInstance().nextId();
            OrderMqProducer.builderOrderMsgList(orderMsgList, orderId);
        }

        for (int i = 0; i < orderMsgList.size(); i++) {
            OrderMsgDTO orderMsgDTO = orderMsgList.get(i);
            String body = orderMsgDTO.toString();
            Message msg = new Message("OrderTopic",
                    tags[i % tags.length],
                    "OrderKey" + i,
                    body.getBytes(RemotingHelper.DEFAULT_CHARSET));

            SendResult sendResult = defaultMQProducer.send(msg, new MessageQueueSelector() {

                //List<MessageQueue> msgQueList 訊息要傳送的Topic下的所有分割槽
                //Message message 訊息物件
                // Object args 額外的引數，使用者可以自己傳遞引數
                // 比如為了把同一個訂單的訊息傳送到同一個分割槽中，
                // 可以把訂單號作為一個引數傳遞過去然後mod分割槽個數，
                // 就可以保證把同一個訂單的訊息傳送到同一個分割槽中去
                @Override
                public MessageQueue select(List<MessageQueue> msgQueList, Message message, Object args) {
                    long orderId = (long) args;
                    long index = orderId % msgQueList.size();
                    return msgQueList.get((int) index);
                }
            }, orderMsgDTO.getOrderId());

            System.out.println(sendResult +
                    String.format("message [%s] send success.",
                            new String(msg.getBody())));
        }
        //defaultMQProducer.shutdown();
    }


    private static void builderOrderMsgList(List<OrderMsgDTO> orderMsgList, long orderId) {
        orderMsgList.add(new OrderMsgDTO(orderId, "Create"));
        orderMsgList.add(new OrderMsgDTO(orderId, "PayOff"));
        orderMsgList.add(new OrderMsgDTO(orderId, "Finish"));
    }
}

2.topic配置資訊

3 執行結果- 5個訂單的訊息分到了三個佇列（queue=0,1,2）中去了

SendResult [sendStatus=SEND_OK, 
            msgId=AC140A051DC818B4AAC2938813480000, 
            offsetMsgId=AC140A3C00002A9F00000000000297A4, 
            messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=0], 
            queueOffset=3
        ]
message [OrderMsgDTO(orderId=517724620921503744, msgType=Create)] send success.
SendResult [sendStatus=SEND_OK, 
            msgId=AC140A051DC818B4AAC2938813510001, 
            offsetMsgId=AC140A3C00002A9F000000000002988D, 
            messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=0], 
            queueOffset=4
        ]
message [OrderMsgDTO(orderId=517724620921503744, msgType=PayOff)] send success.
SendResult [sendStatus=SEND_OK, 
            msgId=AC140A051DC818B4AAC2938813C90002, 
            offsetMsgId=AC140A3C00002A9F0000000000029976, 
            messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=0], 
            queueOffset=5
        ]
message [OrderMsgDTO(orderId=517724620921503744, msgType=Finish)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938813EA0003, offsetMsgId=AC140A3C00002A9F0000000000029A5F, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=1], queueOffset=3]message [OrderMsgDTO(orderId=517724620925698048, msgType=Create)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938813F00004, offsetMsgId=AC140A3C00002A9F0000000000029B48, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=1], queueOffset=4]message [OrderMsgDTO(orderId=517724620925698048, msgType=PayOff)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938814000005, offsetMsgId=AC140A3C00002A9F0000000000029C31, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=1], queueOffset=5]message [OrderMsgDTO(orderId=517724620925698048, msgType=Finish)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938814A20006, offsetMsgId=AC140A3C00002A9F0000000000029D1A, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=2], queueOffset=12]message [OrderMsgDTO(orderId=517724620925698049, msgType=Create)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938815060007, offsetMsgId=AC140A3C00002A9F0000000000029E03, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=2], queueOffset=13]message [OrderMsgDTO(orderId=517724620925698049, msgType=PayOff)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938815110008, offsetMsgId=AC140A3C00002A9F0000000000029EEC, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=2], queueOffset=14]message [OrderMsgDTO(orderId=517724620925698049, msgType=Finish)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938815150009, offsetMsgId=AC140A3C00002A9F0000000000029FD5, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=0], queueOffset=6]message [OrderMsgDTO(orderId=517724620925698050, msgType=Create)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC293881546000A, offsetMsgId=AC140A3C00002A9F000000000002A0BE, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=0], queueOffset=7]message [OrderMsgDTO(orderId=517724620925698050, msgType=PayOff)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC293881549000B, offsetMsgId=AC140A3C00002A9F000000000002A1A8, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=0], queueOffset=8]message [OrderMsgDTO(orderId=517724620925698050, msgType=Finish)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938819F0000C, offsetMsgId=AC140A3C00002A9F000000000002A292, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=1], queueOffset=6]message [OrderMsgDTO(orderId=517724620925698051, msgType=Create)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC293881A30000D, offsetMsgId=AC140A3C00002A9F000000000002A37C, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=1], queueOffset=7]message [OrderMsgDTO(orderId=517724620925698051, msgType=PayOff)] send success.
SendResult [sendStatus=SEND_OK, msgId=AC140A051DC818B4AAC2938821A0000E, offsetMsgId=AC140A3C00002A9F000000000002A466, messageQueue=MessageQueue [topic=OrderTopic, brokerName=broker-a, queueId=1], queueOffset=8]message [OrderMsgDTO(orderId=517724620925698051, msgType=Finish)] send success.

5.消費者程式碼-兩個消費者程式碼相同只是細微的差別

    1.程式碼

package com.roger.order.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

public class OrderMqConsumer1 {

    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer defaultMQPushConsumer =
                new DefaultMQPushConsumer("orderMQPushConsumerGroup");
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        defaultMQPushConsumer.setNamesrvAddr("172.20.10.60:9876");

        defaultMQPushConsumer.subscribe("OrderTopic","*");

        defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
            Random r = new Random();
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgExtList, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                System.out.println("當前執行緒名：" + Thread.currentThread().getName() + ";Receive new message:");
                for(MessageExt msgExt : msgExtList){
                    System.out.println(String.format("Consume message [%s],TagName [%s]",
                            new String(msgExt.getBody()),
                            msgExt.getTags()));
                    try {
                        //簡單業務處理邏輯
                        TimeUnit.SECONDS.sleep(r.nextInt(10));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        defaultMQPushConsumer.start();

        System.out.println("OrderMqConsumer1 Started...");
    }
}

   2.OrderMqConsumer1執行結果-

             (a) 消費了queue=0 和queue=1的資料

             (b) 每個訂單按照順序執行，並且是訂單號相同的訊息同一個執行緒執行的

OrderMqConsumer1 Started...
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698048, msgType=Create)],TagName [TagC]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698048, msgType=PayOff)],TagName [TagP]
當前執行緒名：ConsumeMessageThread_2;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620921503744, msgType=Create)],TagName [TagC]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698048, msgType=Finish)],TagName [TagF]
當前執行緒名：ConsumeMessageThread_2;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620921503744, msgType=PayOff)],TagName [TagP]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698051, msgType=Create)],TagName [TagC]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698051, msgType=PayOff)],TagName [TagP]
當前執行緒名：ConsumeMessageThread_2;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620921503744, msgType=Finish)],TagName [TagF]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698051, msgType=Finish)],TagName [TagF]
當前執行緒名：ConsumeMessageThread_2;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698050, msgType=Create)],TagName [TagC]
當前執行緒名：ConsumeMessageThread_2;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698050, msgType=PayOff)],TagName [TagP]
當前執行緒名：ConsumeMessageThread_2;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698050, msgType=Finish)],TagName [TagF]


       3.OrderMqConsumer1執行結果-

OrderMqConsumer2 Started...
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698049, msgType=Create)],TagName [TagC]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698049, msgType=PayOff)],TagName [TagP]
當前執行緒名：ConsumeMessageThread_1;Receive new message:
Consume message [OrderMsgDTO(orderId=517724620925698049, msgType=Finish)],TagName [TagF]

相關推薦

RocketMQ-順序訊息

github程式碼下載地址 1.RocketMQ有三種訊息     1)普通訊息     2)順序訊息     3)事務訊息 2.順序訊息     概念：是MQ提供的一種嚴格按照順序進行釋出和消費

RocketMQ實現順序訊息

大部分業務場景不需要訊息的產生和消費的順序相同(順序消費)。但在某些場景（比如 訂單的生成，付款 ，下單 這3個訊息必須按順序處理）。解決思路是：（1）在傳送端，要把統一業務ID 傳送到同一個MessageQueue. (2) 在消費端，同一MessageQueue 的訊息不要被併發處理了。

分散式訊息佇列RocketMQ&Kafka -- 訊息的“順序消費”-- 一個看似簡單的複雜問題

在說到訊息中介軟體的時候，我們通常都會談到一個特性：訊息的順序消費問題。這個問題看起來很簡單：Producer傳送訊息1, 2, 3。。。 Consumer按1, 2, 3。。。順序消費。 但實際情況卻是：無論RocketMQ，還是Kafka，預設都不保證訊息

RocketMQ原理解析-producer 3.如何傳送順序訊息

Rocketmq能夠保證訊息嚴格順序，但是Rocketmq需要producer保證順序訊息按順序傳送到同一個queue中，比如購買流程(1)下單(2)支付(3)支付成功，這三個訊息需要根據特定規則將這個三個訊息按順序傳送到一個queue 如何實現把順序訊息傳送到同一個qu

RocketMQ入門到入土（二）事務訊息&順序訊息

接上一篇：RocketMQ入門到入土（一）新手也能看懂的原理和實戰！ 一、事務訊息的由來 1、案例 引用官方的購物案例： 小明購買一個100元的東西，賬戶扣款100元的同時需要保證在下游的積分系統給小明這個賬號增加100積分。賬號系統和積分系統是兩個獨立是系統，一個要減少100元，一個要增加100積分。如下圖

Kafka、RabbitMQ、RocketMQ等訊息中介軟體的對比 —— 訊息傳送效能和區別

分散式系統中,我們廣泛運用訊息中介軟體進行系統間的資料交換,便於非同步解耦。現在開源的訊息中介軟體有很多,前段時間我們自家的產品 RocketMQ (MetaQ的核心) 也順利開源,得到大家的關注。 那麼,訊息中介軟體效能究竟哪家強? 帶著這個疑問,我們中介軟體測

RocketMQ-事務訊息

1.RocketMQ有三種訊息     1)普通訊息     2)順序訊息     3)事務訊息 2.分散式事務-強調最終一致性而不是強一致性 單個應用，資料庫分庫分表了（跨資料庫） 多個應用，服務

解析RocketMQ的訊息索引檔案consumequeue

CommitLog的檔案結構 下圖展示了CommitLog的檔案結構，可以看到，包含了topic、queueId、訊息體等核心資訊。 同Kafka一樣，訊息是變長的，順序寫入。 如下圖所示： ConsumeQueue的檔案結構 ConsumeQueue中並不需要

架構師日記——Kafka、RabbitMQ、RocketMQ等訊息中介軟體的對比

分散式系統中,我們廣泛運用訊息中介軟體進行系統間的資料交換,便於非同步解耦。現在開源的訊息中介軟體有很多,前段時間我們自家的產品 RocketMQ (MetaQ的核心) 也順利開源,得到大家的關注。 那麼,訊息中介軟體效能究竟哪家強? 帶著這個疑問,我們中

RocketMQ的訊息傳送方式

同步傳送 簡單來說，同步傳送就是指 producer 傳送訊息後，會在接收到 broker 響應後才繼續發下一條訊息的通訊方式。 由於這種同步傳送的方式確保了訊息的可靠性，同時也能及時得到訊息傳送的結果，故而適合一些傳送比較重要的訊息場景，比如說重要的通知郵件、營銷簡訊等等。在實

RocketMq-延遲訊息及 程式碼實現

支援延遲訊息 RocketMQ 支援定時訊息，但是不支援任意時間精度，僅支援特定的 level，例如定時 5s， 10s， 1m 等。其中，level=0 級表示不延時，level=1 表示 1 級延時，level=2 表示 2 級延時，以此類推。 配置 開啟安裝目錄的

RocketMQ的訊息佇列

訊息佇列的特徵：1.業務無關。訊息佇列不需要考慮上層的業務模型，只需要做好訊息分發；2.FIFO。先投遞先到達的保證是一個訊息佇列和一個buffer（快取）的本質區別；3.容災。主要包括節點的動態增刪和訊息的持久化。4.效能。訊息佇列的吞吐量提升，則整個系統的內部通訊效率就會提高。為什麼需要訊息佇列？當系統中

rocketmq 順序消費理解

要實現佇列的順序消費，比如(1)下單(2)支付(3)支付 預設的傳送會隨機指定一個佇列， SendResult result = producer.send(msg); public MessageQueue selectOneMessageQueue(final T

搞懂分散式技術19：使用RocketMQ事務訊息解決分散式事務

說到分散式事務，就會談到那個經典的”賬號轉賬”問題：2個賬號，分佈處於2個不同的DB，或者說2個不同的子系統裡面，A要扣錢，B要加錢，如何保證原子性？一般的思路都是通過訊息中介軟體來實現“最終一致性”：A系統扣錢，然後發條訊息給中介軟體，B系統接收此訊息，進行加錢。但這裡面有個問題：A是先update DB，

分散式訊息佇列RocketMQ--事務訊息--解決分散式事務

說到分散式事務，就會談到那個經典的”賬號轉賬”問題：2個賬號，分佈處於2個不同的DB，或者說2個不同的子系統裡面，A要扣錢，B要加錢，如何保證原子性？ 一般的思路都是通過訊息中介軟體來實現“最終一致性”：A系統扣錢，然後發條訊息給中介軟體，B系統接收此訊息，進行加錢。 但這裡面有個問題：A是先update D

RocketMQ 解決訊息重複投遞的問題

業務與訊息耦合的情況下會產生訊息重複投遞，因為支付寶儲存了訊息記錄，餘額寶處理成功之後需要傳送處理成功的訊息給支付寶。支付寶進行確認之後，把訊息記錄狀態刪除或者做變更。而我們的業務與訊息解耦的方式則不需要考慮訊息重複投遞的問題。因為 我們完全信任MQ中介軟體。第一種方式的訊息

RocketMQ原始碼分析之RocketMQ事務訊息實現原理上篇(二階段提交)

根據上文的描述，傳送事務訊息的入口為： TransactionMQProducer#sendMessageInTransaction： public TransactionSendResult sendMessageInTransaction(final Message msg, final Object

RocketMq——順序消費和事務

RocketMQ不遵循JMS規範，自己有一套自定義的機制，使用訂閱主題的方式傳送和接收任務，支援廣播和叢集兩種消費模式。 叢集模式：設定消費端物件屬性：MessageModel.CLUSTERING,這種方式就可以達到ActiveMQ水平擴充套件負載均衡消費訊

RocketMQ——順序消費（程式碼）

本部落格主要是以程式碼示例來了解順序消費的相關內容,建議在此之前先了解下順序消費的原理。注：RocketMQ可以嚴格的保證訊息有序，但這個順序，不是全域性順序，只是分割槽（queue）順序，如果想要全域性順序，那麼需要保證只有一個分割槽。順序消費簡介1.普通順序消費順序消費的

RocketMQ原始碼分析之從官方示例窺探：RocketMQ事務訊息實現基本思想

RocketMQ4.3.0版本開始支援事務訊息，後續分享將開始將剖析事務訊息的實現原理。首先從官方給出的Demo例項入手，以此通往RocketMQ事務訊息的世界中。 官方版本未釋出之前，從apache rocketmq第一個版本上線後，程式碼中存在與事務訊息相關的程式碼，例如COMMIT、ROLLBACK、