消息隊(duì)列技術(shù)選型:這 7 種消息場(chǎng)景一定要考慮!

大家好，我是小富～

我們?cè)谧鱿㈥?duì)列的技術(shù)選型時(shí)，往往會(huì)結(jié)合業(yè)務(wù)場(chǎng)景進(jìn)行考慮。今天來(lái)聊一聊消息隊(duì)列可能會(huì)用到的 7 種消息場(chǎng)景。

1 普通消息

消息隊(duì)列最基礎(chǔ)的功能就是生產(chǎn)者發(fā)送消息、Broker 保存消息，消費(fèi)者來(lái)消費(fèi)消息，以此實(shí)現(xiàn)系統(tǒng)解耦、削峰填谷的作用。

普通消息是消息隊(duì)列必備的消息類型，也是系統(tǒng)使用場(chǎng)景最多的一種消息。

2 順序消息

順序消息是指生產(chǎn)者發(fā)送消息的順序和消費(fèi)者消費(fèi)消息的順序是一致的。比如在一個(gè)電商場(chǎng)景，同一個(gè)用戶提交訂單、訂單支付、訂單出庫(kù)，這三個(gè)消息消費(fèi)者需要按照順序來(lái)進(jìn)行消費(fèi)。如下圖：

順序消息的實(shí)現(xiàn)并不容易，原因如下：

• 生產(chǎn)者集群中，有多個(gè)生產(chǎn)者發(fā)送消息，網(wǎng)絡(luò)延遲不一樣，很難保證發(fā)送到 Broker 的消息落盤順序是一致的；
• 如果 Broker 有多個(gè)分區(qū)或隊(duì)列，生產(chǎn)者發(fā)送的消息會(huì)進(jìn)入多個(gè)分區(qū)，也無(wú)法保證順序消費(fèi)；
• 如果有多個(gè)消費(fèi)者來(lái)異步消費(fèi)同一個(gè)分區(qū)，很難保證消費(fèi)順序跟生產(chǎn)者發(fā)送順序一致。

要保證消息有序，需要滿足兩個(gè)條件：

• 同一個(gè)生產(chǎn)者必須同步發(fā)送消息到同一個(gè)分區(qū)；
• 一個(gè)分區(qū)只能給同一個(gè)消費(fèi)者消費(fèi)。

如下圖：

上面第二個(gè)條件是比較容易實(shí)現(xiàn)的，一個(gè)分區(qū)綁定一個(gè)消費(fèi)者就可以，主要是第一個(gè)條件。

在主流消息隊(duì)列的實(shí)現(xiàn)中，Kafka 和 Pulsar 的實(shí)現(xiàn)方式類似，生產(chǎn)者給消息賦值一個(gè) key，對(duì) key 做 Hash 運(yùn)算來(lái)指定消息發(fā)送到哪一個(gè)分區(qū)。比如上面電商的例子，對(duì)同一個(gè)用戶的一筆訂單，提交訂單、訂單支付、訂單出庫(kù)這三個(gè)消息賦值同一個(gè) key，就可以把這三條消息發(fā)送到同一個(gè)分區(qū)。

對(duì)于 RocketMQ，生產(chǎn)者在發(fā)送消息的時(shí)候，可以通過(guò) MessageQueueSelector 指定把消息投遞到那個(gè) MessageQueue，如下圖：

示例代碼如下：

      
      public static void main(String[] args) throws UnsupportedEncodingException {
 try {
  DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
  producer.start();

  String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
  for (int i = 0; i < 100; i++) {
   int orderId = i % 10;
   Message msg =
    new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
     ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
   SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
     Integer id = (Integer) arg;
     int index = id % mqs.size();
     return mqs.get(index);
    }
   }, orderId);

   System.out.printf("%s%n", sendResult);
  }

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

    

RabbitMQ 的實(shí)現(xiàn)是 Exchange 根據(jù)設(shè)置好的 Route Key 將數(shù)據(jù)路由到不同的 Queue 中。示例代碼如下：

      
      @Resource
private AmqpTemplate rabbitTemplate;

public void send1(String message) {
 rabbitTemplate.convertAndSend("testExchange", "testRoutingKey", message);
}

    

3 延時(shí)消息

或者也叫定時(shí)消息，是指消息發(fā)送后不會(huì)立即被消費(fèi)，而是指定一個(gè)時(shí)間，到時(shí)間后再消費(fèi)。經(jīng)典的場(chǎng)景比如電商購(gòu)物時(shí)，30 分鐘未支付訂單，讓訂單自動(dòng)失效。

3.1 RocketMQ 實(shí)現(xiàn)

RocketMQ 定義了 18 個(gè)延時(shí)級(jí)別，每個(gè)延時(shí)級(jí)別對(duì)應(yīng)一個(gè)延時(shí)時(shí)間。下面如果延遲級(jí)別是 3，則消息會(huì)延遲 10s 才會(huì)拉取。

      
      //MessageStoreConfig類
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

    

RocketMQ 的延時(shí)消息如下圖：

生產(chǎn)者把消費(fèi)發(fā)送到 Broker 后，Broker 首先把消息保存到 SCHEDULE_TOPIC_XXXX 這個(gè) Topic，然后調(diào)度任務(wù)會(huì)判斷是否到期，如果到期，會(huì)把消息從 SCHEDULE_TOPIC_XXXX 取出投遞到原始的 queue，這樣消費(fèi)者就可以消費(fèi)到了。

RocketMQ 的延時(shí)消息只支持最大兩個(gè)小時(shí)的延時(shí)，不過(guò) RocketMQ5.0 基于時(shí)間輪算法實(shí)現(xiàn)了定時(shí)消息，解決了這個(gè)問(wèn)題。

3.2 Pulsar 實(shí)現(xiàn)

Pulsar 的實(shí)現(xiàn)如下圖：

Pulsar 的延時(shí)消息首先會(huì)寫入一個(gè) Delayed Message Tracker 的數(shù)據(jù)結(jié)構(gòu)中，Delayed Message Tracker 根據(jù)延時(shí)時(shí)間構(gòu)建 delayed index 優(yōu)先級(jí)隊(duì)列。消費(fèi)者拉取消息時(shí)，首先去 Delayed Message Tracker 檢查是否有到期的消息。如果有則直接拉取進(jìn)行消費(fèi)。

3.3 RabbitMQ 實(shí)現(xiàn)

RabbitMQ 的實(shí)現(xiàn)方式有兩種，一種是投遞到普通隊(duì)列都不消費(fèi)，等消息過(guò)期后被投遞到死信隊(duì)列，消費(fèi)者消費(fèi)死信隊(duì)列。如下圖：

第二種方式是生產(chǎn)者發(fā)送消息時(shí)，先發(fā)送到本地 Mnesia 數(shù)據(jù)庫(kù)，消息到期后定時(shí)器再將消息投遞到 broker。

3.4 Kafka 實(shí)現(xiàn)

Kafka 本身并沒(méi)有延時(shí)隊(duì)列，不過(guò)可以通過(guò)生產(chǎn)者攔截器來(lái)實(shí)現(xiàn)消息延時(shí)發(fā)送，也可以定義延時(shí) Topic，利用類似 RocketMQ 的方案來(lái)實(shí)現(xiàn)延時(shí)消息。

4 事務(wù)消息

事務(wù)消息是指生產(chǎn)消息和消費(fèi)消息滿足事務(wù)的特性。

RabbitMQ 和 Kafka 的事務(wù)消息都是只支持生產(chǎn)消息的事務(wù)特性，即一批消息要不全部發(fā)送成功，要不全部發(fā)送失敗。

RabbitMQ 通過(guò) Channel 來(lái)開(kāi)啟事務(wù)消息，代碼如下：

      
      ConnectionFactory factory=new ConnectionFactory();
connection=factory.newConnection();
Channel channel=connection.createChannel();
//開(kāi)啟事務(wù)
channel.txSelect();
channel.basicPublish("directTransactionExchange","transactionRoutingKey",null,message.getBytes("utf-8"));
//提交事務(wù) 或者 channel.txRollback()回滾事務(wù)
channel.txCommit();

    

Kafka 可以給多個(gè)生產(chǎn)者設(shè)置同一個(gè)事務(wù) ID ，從而把多個(gè) Topic 、多個(gè) Partition 放在一個(gè)事務(wù)中，實(shí)現(xiàn)原子性寫入。

Pulsar 的事務(wù)消息對(duì)于事務(wù)語(yǔ)義的定義是：允許事件流應(yīng)用將消費(fèi)、處理、生產(chǎn)消息整個(gè)過(guò)程定義為一個(gè)原子操作?？梢?jiàn)，Pulsar 的事務(wù)消息可以覆蓋消息流整個(gè)過(guò)程。

RocketMQ 的事務(wù)消息是通過(guò) half 消息來(lái)實(shí)現(xiàn)的。以電商購(gòu)物場(chǎng)景來(lái)看，賬戶服務(wù)扣減賬戶金額后，發(fā)送消息給 Broker，庫(kù)存服務(wù)來(lái)消費(fèi)這條消息進(jìn)行扣減庫(kù)存。如下圖:

可見(jiàn)，RocketMQ 只能保證生產(chǎn)者發(fā)送消息和本地事務(wù)的原子性，并不能保證消費(fèi)消息的原子性。

5 軌跡消息

軌跡消息主要用于跟蹤消息的生命周期，當(dāng)消息丟失時(shí)可以很方便地找出原因。

軌跡消息也跟普通消息一樣，也需要存儲(chǔ)和查詢，也會(huì)占用消息隊(duì)列的資源，所以選擇軌跡消息要考慮下面幾點(diǎn)：

• 消息生命周期的關(guān)鍵節(jié)點(diǎn)一定要記錄；
• 不能影響正常消息的發(fā)送和消費(fèi)性能；
• 不能影響 Broker 的消息存儲(chǔ)性能；
• 要考慮消息查詢維度和性能。

RabbitMQ Broker 實(shí)現(xiàn)了軌跡消息的功能，打開(kāi) Trace 開(kāi)關(guān)，就可以把軌跡消息發(fā)送到 amq.rabbitmq.trace 這個(gè) exchange，但是要考慮軌跡消息會(huì)不會(huì)給 Broker 造成 壓力進(jìn)而導(dǎo)致消息積壓。RabbitMQ 的生產(chǎn)者和消費(fèi)者都沒(méi)有實(shí)現(xiàn)軌跡消息，需要開(kāi)發(fā)者自己來(lái)實(shí)現(xiàn)。

RocketMQ 生產(chǎn)者、Broker 和消費(fèi)者都實(shí)現(xiàn)了軌跡消息，不過(guò)默認(rèn)是關(guān)閉的，需要手工開(kāi)啟。

使用軌跡消息，需要考慮記錄哪些節(jié)點(diǎn)、存儲(chǔ)介質(zhì)、性能、查詢方式等問(wèn)題。

6 死信隊(duì)列

在消息隊(duì)列中，死信隊(duì)列主要應(yīng)對(duì)一些異常的情況，如下圖：

RocketMQ 實(shí)現(xiàn)了消費(fèi)端的死信隊(duì)列，當(dāng)消費(fèi)者消費(fèi)失敗時(shí)，會(huì)進(jìn)行重試，如果重試 16 次還是失敗，則這條消息會(huì)被發(fā)送到死信隊(duì)列。

RabbitMQ 實(shí)現(xiàn)了生產(chǎn)者和 Broker 的死信隊(duì)列，下面三種情況，消息會(huì)被發(fā)送到死信隊(duì)列：

• 生產(chǎn)者發(fā)送消息被拒絕，并且 requeue 參數(shù)設(shè)置為 false；
• Broker 消息過(guò)期了；
• 隊(duì)列達(dá)到最大長(zhǎng)度。

RabbitMQ 消息變成死信消息后，會(huì)被發(fā)送到死信交換機(jī)（Dead-Letter-Exchange）。

7 優(yōu)先級(jí)消息

有一些業(yè)務(wù)場(chǎng)景下，我們需要優(yōu)先處理一些消息，比如銀行里面的金卡客戶、銀卡客戶優(yōu)先級(jí)高于普通客戶，他們的業(yè)務(wù)需要優(yōu)先處理。如下圖：

主流消息隊(duì)列中，RabbitMQ 是支持優(yōu)先級(jí)隊(duì)列的，代碼如下：

      
      ConnectionFactory factory=new ConnectionFactory();
connection=factory.newConnection();
Channel channel=connection.createChannel();
Map<String, Object> args = new HashMap<String, Object>();
//設(shè)置優(yōu)先級(jí)為 5
args.put("x-max-priority", 5);
channel.queueDeclare("my-priority-queue", true, false, false, args);

    

8 總結(jié)

消息隊(duì)列技術(shù)選型，要考慮的因素很多，本文主要從業(yè)務(wù)場(chǎng)景來(lái)分析需要考慮的因素，同時(shí)技術(shù)上也需要考慮運(yùn)維復(fù)雜度、業(yè)務(wù)規(guī)模、社區(qū)活躍度、學(xué)習(xí)成本等因素。希望本文對(duì)你使用消息隊(duì)列有所幫助。