2. 程式人生 > >RocketMQ原理學習--訊息型別

RocketMQ原理學習--訊息型別

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

一、叢集訊息與廣播訊息

叢集消費：當使用叢集消費模式時，MQ 認為任意一條訊息只需要被叢集內的任意一個消費者處理即可。
廣播消費：當使用廣播消費模式時，MQ 會將每條訊息推送給叢集內所有註冊過的客戶端，保證訊息至少被每臺機器消費一次。

叢集消費模式：

適用場景&注意事項

  • 消費端叢集化部署，每條訊息只需要被處理一次。
  • 由於消費進度在服務端維護，可靠性更高。
  • 叢集消費模式下，每一條訊息都只會被分發到一臺機器上處理，如果需要被叢集下的每一臺機器都處理，請使用廣播模式。
  • 叢集消費模式下，不保證訊息的每一次失敗重投等邏輯都能路由到同一臺機器上，因此處理訊息時不應該做任何確定性假設。

廣播消費模式：

適用場景&注意事項

  • 順序訊息暫不支援廣播消費模式。
  • 每條訊息都需要被相同邏輯的多臺機器處理。
  • 消費進度在客戶端維護，出現重複的概率稍大於叢集模式。
  • 廣播模式下，MQ 保證每條訊息至少被每臺客戶端消費一次，但是並不會對消費失敗的訊息進行失敗重投，因此業務方需要關注消費失敗的情況。
  • 廣播模式下，第一次啟動時預設從最新訊息消費，客戶端的消費進度是被持久化在客戶端本地的隱藏檔案中，因此不建議刪除該隱藏檔案，否則會丟失部分訊息。
  • 廣播模式下，每條訊息都會被大量的客戶端重複處理，因此推薦儘可能使用叢集模式。
  • 廣播模式下服務端不維護消費進度，所以 MQ 控制檯不支援訊息堆積查詢和堆積報警功能。

程式碼示例：

設定叢集訊息：consumer.setMessageModel(MessageModel.CLUSTERING);

設定廣播訊息：consumer.setMessageModel(MessageModel.BROADCASTING);

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
        consumer.setNamesrvAddr("localhost:9876");

        //叢集消費者
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        //廣播消費者
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("TopicA-test", "TagA");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

二、普通訊息、事物訊息、順序訊息、延時訊息

RocketMQ 針對不同的業務場景還提供了普通訊息、事物訊息、順序訊息和延時訊息等幾種訊息型別。

1、普通訊息

普通訊息也叫做無序訊息，簡單來說就是沒有順序的訊息，producer 只管傳送訊息，consumer 只管接收訊息，至於訊息和訊息之間的順序並沒有保證，可能先發送的訊息先消費，也可能先發送的訊息後消費。

舉個簡單例子，producer 依次傳送 order id 為 1、2、3 的訊息到 broker，consumer 接到的訊息順序有可能是 1、2、3，也有可能是 2、1、3 等情況，這就是普通訊息。

因為不需要保證訊息的順序，所以訊息可以大規模併發地傳送和消費，吞吐量很高，適合大部分場景。

示例：

生產者

public class Producer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("rmq-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        try {
            for (int i = 0; i < 3; i++) {
                Message msg = new Message("TopicA-test",// topic
                        "TagA",// tag
                        (new Date() + "Hello RocketMQ ,QuickStart 11" + i)
                                .getBytes()// body
                );
                SendResult sendResult = producer.send(msg);
                System.out.println(sendResult);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
       producer.shutdown();
    }

}

消費者

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
        consumer.setNamesrvAddr("localhost:9876");
        //consumer.setInstanceName("rmq-instance2");
        consumer.subscribe("TopicA-test", "TagA");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

2、事物訊息

MQ 的事務訊息互動流程如下圖所示：

MQ 事務訊息互動流程

採用2PC提交：

第一階段是：步驟1，2，3。
第二階段是：步驟4，5。

生產者：

public class TransactionProducer {

    public static void main(String [] args) throws Exception{

        final TransactionMQProducer producer = new TransactionMQProducer("rmq-transaction");
        producer.setNamesrvAddr("localhost:9876");

        //事務回查最小併發數

        producer.setCheckThreadPoolMinSize(5);

        //事務回查最大併發數

        producer.setCheckThreadPoolMaxSize(20);

        //佇列數

        producer.setCheckRequestHoldMax(2000);

        producer.start();

        //伺服器回撥producer,檢查本地事務分支成功還是失敗
        producer.setTransactionCheckListener(new TransactionCheckListener() {

            @Override
            public LocalTransactionState checkLocalTransactionState(MessageExt messageExt) {

                System.out.println("state --" + new String(messageExt.getBody()));

                return LocalTransactionState.COMMIT_MESSAGE;

            }

        });

        TransactionExecuterImpl transactionExecuter = new TransactionExecuterImpl();

        for (int i = 0; i < 2; i++) {

            Message msg = new Message("TopicTransaction",

                    "Transaction" + i,

                    ("Hello RocketMq" + i).getBytes()

            );

            SendResult sendResult = producer.sendMessageInTransaction(msg, transactionExecuter, "tq");

            System.out.println(sendResult);

            TimeUnit.MICROSECONDS.sleep(1000);

        }

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {

            @Override
            public void run() {

                producer.shutdown();

            }

        }));

        System.exit(0);

    }
}

執行本地事物：

public class TransactionExecuterImpl implements LocalTransactionExecuter {

    @Override
    public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {

        System.out.println("msg=" + new String(msg.getBody()));

        System.out.println("arg = "+arg);

        String tag = msg.getTags();

        if (tag.equals("Transaction1")){

            //這裡有一個分階段提交的概念

            System.out.println("這裡是處理業務邏輯，失敗情況下進行ROLLBACK");

            return LocalTransactionState.ROLLBACK_MESSAGE;

        }

        return LocalTransactionState.COMMIT_MESSAGE;

        //return LocalTransactionState.UNKNOW;

    }

}

消費者：

public class TransactionConsumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-transaction");
        consumer.setNamesrvAddr("localhost:9876");
        //consumer.setInstanceName("rmq-instance2");
        consumer.subscribe("TopicTransaction", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

3、順序訊息

有序訊息就是按照一定的先後順序的訊息型別。

舉個例子來說，producer 依次傳送 order id 為 1、2、3 的訊息到 broker，consumer 接到的訊息順序也就是 1、2、3 ，而不會出現普通訊息那樣的 2、1、3 等情況。

那麼有序訊息是如何保證的呢？我們都知道訊息首先由 producer 到 broker，再從 broker 到 consumer，分這兩步走。那麼要保證訊息的有序，勢必這兩步都是要保證有序的，即要保證訊息是按有序傳送到 broker，broker 也是有序將訊息投遞給 consumer，兩個條件必須同時滿足，缺一不可。
進一步還可以將有序訊息分成

  • 全域性有序訊息
  • 區域性有序訊息

實現原理：由於生產者預設是輪詢獲取MessageQueue佇列(每個Topic預設初始化4個MessageQueue)，然後將訊息輪詢傳送到不同的MessageQueue中，訊息者從MessageQueue中獲取資料時很可能是無序的。

區域性有序訊息：將相同順序的訊息傳送到同一個MessageQueue佇列，這樣消費者從佇列中獲取資料肯定是相對有序的。

全域性有序訊息：將所有的訊息傳送到一個MessageQueue佇列，消費者從單個佇列中拉取訊息，訊息有序。

生產者：實現MessageQueueSelector介面，相同順序的訊息獲取同一個MessageQueue

public class OrderProducer {

    public static void main(String[] args) throws Exception {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("order_Producer");
            producer.setNamesrvAddr("localhost:9876");

            producer.start();

            for (int i = 1; i <= 5; i++) {

                Message msg = new Message("TopicOrderTest", "order_1", "KEY" + i, ("order_1 " + i).getBytes());

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, 0);

                System.out.println(sendResult);
            }
            for (int i = 1; i <= 5; i++) {

                Message msg = new Message("TopicOrderTest", "order_2", "KEY" + i, ("order_2 " + i).getBytes());

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, 1);

                System.out.println(sendResult);
            }
            for (int i = 1; i <= 5; i++) {

                Message msg = new Message("TopicOrderTest", "order_3", "KEY" + i, ("order_3 " + i).getBytes());

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, 2);

                System.out.println(sendResult);
            }

            producer.shutdown();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消費者：設定訊息監聽器為順序訊息監聽器MessageListenerOrderly

public class OrderConsumer {


    public static void main(String[] args) throws MQClientException {


        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_Consumer");
        consumer.setNamesrvAddr("localhost:9876");

        /**
         * 設定Consumer第一次啟動是從佇列頭部開始消費還是佇列尾部開始消費<br>
         * 如果非第一次啟動，那麼按照上次消費的位置繼續消費
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("TopicOrderTest", "*");

        /**
         * 實現了MessageListenerOrderly表示一個佇列只會被一個執行緒取到
         *，第二個執行緒無法訪問這個佇列
         */
        consumer.registerMessageListener(new MessageListenerOrderly() {
            AtomicLong consumeTimes = new AtomicLong(0);

            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                // 設定自動提交
                context.setAutoCommit(true);
                for (MessageExt msg : msgs) {
                    System.out.println(msg + ",內容：" + new String(msg.getBody()));
                }

                try {
                    TimeUnit.SECONDS.sleep(5L);
                } catch (InterruptedException e) {

                    e.printStackTrace();
                }
                ;

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();

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


}

4、延時訊息

延時訊息，簡單來說就是當 producer 將訊息傳送到 broker 後，會延時一定時間後才投遞給 consumer 進行消費。

RcoketMQ的延時等級為：1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h。level=0，表示不延時。level=1，表示 1 級延時，對應延時 1s。level=2 表示 2 級延時，對應5s，以此類推。

這種訊息一般適用於訊息生產和消費之間有時間視窗要求的場景。比如說我們網購時，下單之後是有一個支付時間，超過這個時間未支付，系統就應該自動關閉該筆訂單。那麼在訂單建立的時候就會就需要傳送一條延時訊息（延時15分鐘）後投遞給 consumer，consumer 接收訊息後再對訂單的支付狀態進行判斷是否關閉訂單。

設定延時非常簡單，只需要在Message設定對應的延時級別即可

生產者：

public class DelayProducer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("rmq-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        try {
            for (int i = 0; i < 3; i++) {
                Message msg = new Message("TopicA-test",// topic
                        "TagA",// tag
                        (new Date() + "Hello RocketMQ ,QuickStart 11" + i)
                                .getBytes()// body
                );
                //1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h。
                // level=0，表示不延時。level=1，表示 1 級延時，對應延時 1s。level=2 表示 2 級延時，對應5s，以此類推
                msg.setDelayTimeLevel(2);

                SendResult sendResult = producer.send(msg);
                System.out.println(sendResult);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
       producer.shutdown();
    }

}

 

 

參考部落格：

https://my.oschina.net/xinxingegeya/blog/1577410

https://help.aliyun.com/document_detail/29548.html?spm=a2c4g.11186623.6.575.42512de7YsiiZ5

https://www.jianshu.com/p/11e875074a8f

相關推薦

RocketMQ原理學習--訊息型別

一、叢集訊息與廣播訊息 叢集消費：當使用叢集消費模式時，MQ 認為任意一條訊息只需要被叢集內的任意一個消費者處理即可。 廣播消費：當使用廣播消費模式時，MQ 會將每條訊息推送給叢集內所有註冊過的客戶端，保證訊息至少被每臺機器消費一次。 叢集消費模式： 適用場景&注意事項

RocketMQ原理學習---Producer訊息傳送

        上一篇部落格RocketMQ原理學習-- Name Server中我們介紹了Name Server提供的相關功能，這篇部落格我們來介紹一下生產者訊息傳送相關的內容。 訊息傳送示例： public class Producer {

RocketMQ原理學習--消費者消費訊息

        在之前的一篇部落格《RocketMQ原理學習--訊息型別》中我們有介紹過RocketMQ的訊息型別，這篇部落格我們簡單介紹一下RocketMQ消費者是如何消費訊息的。 一、Pull or Push     &nb

RocketMQ原理學習--失敗訊息實現原理

        上一篇部落格《 RocketMQ原理學習--延時訊息實現原理》我們介紹了延時訊息的實現原理，這篇部落格我們看看RocketMQ在消費者訊息消費失敗的情況下是如何處理的。         正常情況下

RocketMQ原理學習--延時訊息實現原理

         RocketMQ提供了延時訊息型別，簡單來說就是生產者在傳送訊息的時候指定一個延時時間，當到達延時時間之後訊息才能夠被投送到消費者。       首先我們可以考慮一下RocketMQ的延時訊息是

RocketMQ原理學習---Broker訊息接收處理

         上一篇部落格《RocketMQ原理學習---Producer訊息傳送》中我們簡單瞭解了RocketMQ生產者訊息傳送的過程，接下來我們看看Broker是如何處理接收到的訊息。       &nb

RocketMQ原理學習---生產者事物訊息傳送

        上一篇部落格《RocketMQ原理學習---生產者普通訊息傳送》我們已經對生產者傳送普通訊息有了簡單的瞭解，這篇部落格我們來學習一下RocketMQ在傳送事物訊息時做了什麼處理操作。 一、生產者傳送訊息    

RocketMQ原理學習--死信訊息實現原理

        上一篇部落格《RocketMQ原理學習--失敗訊息實現原理》中我們瞭解到RocketMQ對於失敗訊息的處理原理，當訊息一直失敗的情況下RocketMQ是如何處理的，這篇部落格我們通過分析原始碼簡單瞭解一下。    

RocketMQ原理學習-- Name Server

       Name Server作為RocketMQ的一個元件，其作用就是一個註冊中心，用於管理Broker相關的一些資訊，生產者和消費者可以從Name Server中獲取Broker中相關的Topic資訊等，Name Server可以單臺部署也可以多

RocketMQ原理學習--RocketMQ整體架構窺探

 RocketMQ總體架構圖： RocketMQ簡單架構 角色簡介： 1、Name Server：簡單可以理解為註冊中心，Broker相關資訊會註冊到Name Server叢集中，維護Broker及Broker中相關Top

RocketMQ原理學習--RocketMQ原始碼執行

    最近打算對RocketMQ相關的知識和原始碼進行學習一下，首先能把原始碼匯入及執行能比較方便我們通過跟蹤原始碼進行相關知識學習。 一、工程匯入 git地址:https://github.com/apache/rocketmq.git 直接以maven工程的

RocketMQ原理學習--索引

  1、RocketMQ原理學習--RocketMQ原始碼執行   2、RocketMQ原理學習--RocketMQ整體架構窺探   3、RocketMQ原理學習--訊息型別   4、RocketMQ原理學習-- Name Se

必知必會的RocketMQ訊息型別

普通訊息 普通訊息也叫做無序訊息，簡單來說就是沒有順序的訊息，producer 只管傳送訊息，consumer 只管接收訊息，至於訊息和訊息之間的順序並沒有保證，可能先發送的訊息先消費，也可能先發送的訊息後消費。 舉個簡單例子，producer 依次傳送 order id 為 1、2、3 的

RocketMQ及分散式訊息系統的原理以及重要問題解讀

看到一篇好文章，記錄一下：http://www.jianshu.com/p/453c6e7ff81c 先總結下他的解決方案： 把訊息順序放到 業務層去實現，同時解決訊息重複的問題（因為其最終目標是要叢集的高容錯性和高吞吐量，因為為了保證順序，mq server必定要與每一個

RocketMQ原理（4）——訊息ACK機制及消費進度管理

https://zhuanlan.zhihu.com/p/25265380 consumer的每個例項是靠佇列分配來決定如何消費訊息的。那麼消費進度具體是如何管理的，又是如何保證訊息成功消費的（RocketMQ有保證訊息肯定消費成功的特性（失敗則重試）？ 本文將詳細解

RocketMQ原理解析-consumer 5.push消費-順序消費訊息

順序消費服務ConsumeMessageConcurrentlyService構建的時候                    構建一個執行緒池來接收消費請求ConsumeRequest                    構建一個單執行緒的本地執行緒,用來稍後定時重新消

RocketMQ原理解析-producer 2.如何傳送訊息

Producer輪詢某topic下的所有佇列的方式來實現傳送方的負載均衡 1）  Topic下的所有佇列如何理解： 比如broker1, broker2, borker3三臺broker機器都配置了Topic_A Broker1 的佇列為queue0 , queue1

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

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

RocketMQ學習(七)——RocketMQ延時訊息

使用者傳送的訊息也可以指定延時時間（更準確的說是延時等級），然後在指定延時時間之後投遞訊息，然後被consumer消費。阿里雲的ons還支援定時訊息，而且延時訊息是直接指定延時時間，其實阿里雲的延時訊息也是定時訊息的另一種表述方式，都是通過設定訊息被投遞的時間

[分散式學習]訊息佇列之rocketmq筆記

文件地址 RocketMQ架構 嗶哩嗶哩上的視訊 mq有很多，近期買了《分散式訊息中介軟體實踐》這本書，學習關於mq的相關知識。mq大致有有4個功能： 非同步處理。比如業務端需要給使用者傳送郵件，不需要等傳送完了之後才讓業務端的呼叫程式碼返回。 服務解耦。服務之間呼叫不需要在程式碼上寫死呼叫某某服務，只