一、普通消息

1 消息發(fā)送分類

Producer對(duì)于消息的發(fā)送方式也有多種選擇，不同的方式會(huì)產(chǎn)生不同的系統(tǒng)效果。

同步發(fā)送消息

同步發(fā)送消息是指，Producer發(fā)出?條消息后，會(huì)在收到MQ返回的ACK之后才發(fā)下?條消息。該方式 的消息可靠性最高，但消息發(fā)送效率太低。

異步發(fā)送消息

異步發(fā)送消息是指，Producer發(fā)出消息后無需等待MQ返回ACK，直接發(fā)送下?條消息。該方式的消息可靠性可以得到保障，消息發(fā)送效率也可以。

單向發(fā)送消息

單向發(fā)送消息是指，Producer僅負(fù)責(zé)發(fā)送消息，不等待、不處理MQ的ACK。該發(fā)送方式時(shí)MQ也不返回ACK。該方式的消息發(fā)送效率最高，但消息可靠性較差。

二、順序消息

1 什么是順序消息

順序消息指的是，嚴(yán)格按照消息的發(fā)送順序進(jìn)行消費(fèi)的消息(FIFO)。

默認(rèn)情況下生產(chǎn)者會(huì)把消息以Round Robin輪詢方式發(fā)送到不同的Queue分區(qū)隊(duì)列；而消費(fèi)消息時(shí)會(huì)從多個(gè)Queue上拉取消息，這種情況下的發(fā)送和消費(fèi)是不能保證順序的。如果將消息僅發(fā)送到同一個(gè)Queue中，消費(fèi)時(shí)也只從這個(gè)Queue上拉取消息，就嚴(yán)格保證了消息的順序性。

2 為什么需要順序消息

例如，現(xiàn)在有TOPIC ORDER_STATUS (訂單狀態(tài))，其下有4個(gè)Queue隊(duì)列，該Topic中的不同消息用于 描述當(dāng)前訂單的不同狀態(tài)。假設(shè)訂單有狀態(tài)：未支付、已支付、發(fā)貨中、發(fā)貨成功、發(fā)貨失敗。

根據(jù)以上訂單狀態(tài)，生產(chǎn)者從時(shí)序上可以生成如下幾個(gè)消息：

訂單T0000001:未支付 --> 訂單T0000001:已支付 --> 訂單T0000001:發(fā)貨中 --> 訂單

T0000001:發(fā)貨失敗

消息發(fā)送到MQ中之后，Queue的選擇如果采用輪詢策略，消息在MQ的存儲(chǔ)可能如下：

這種情況下，我們希望Consumer消費(fèi)消息的順序和我們發(fā)送是一致的，然而上述MQ的投遞和消費(fèi)方式，我們無法保證順序是正確的。對(duì)于順序異常的消息，Consumer即使設(shè)置有一定的狀態(tài)容錯(cuò)，也不能完全處理好這么多種隨機(jī)出現(xiàn)組合情況。

基于上述的情況，可以設(shè)計(jì)如下方案：對(duì)于相同訂單號(hào)的消息，通過一定的策略，將其放置在一個(gè)Queue中，然后消費(fèi)者再采用一定的策略（例如，一個(gè)線程獨(dú)立處理一個(gè)queue，保證處理消息的順序性），能夠保證消費(fèi)的順序性。

3 有序性分類

根據(jù)有序范圍的不同，RocketMQ可以嚴(yán)格地保證兩種消息的有序性：分區(qū)有序與全局有序。

全局有序

當(dāng)發(fā)送和消費(fèi)參與的Queue只有一個(gè)時(shí)所保證的有序是整個(gè)Topic中消息的順序， 稱為全局有序。

在創(chuàng)建Topic時(shí)指定Queue的數(shù)量。有三種指定方式：

1）在代碼中創(chuàng)建Producer時(shí)，可以指定其自動(dòng)創(chuàng)建的Topic的Queue數(shù)量

2）在RocketMQ可視化控制臺(tái)中手動(dòng)創(chuàng)建Topic時(shí)指定Queue數(shù)量

3）使用mqadmin命令手動(dòng)創(chuàng)建Topic時(shí)指定Queue數(shù)量

分區(qū)有序

如 果有多個(gè)Queue參與，其僅可保證在該Queue分區(qū)隊(duì)列上的消息順序，則稱為分區(qū)有序。

如何實(shí)現(xiàn)Queue的選擇？在定義Producer時(shí)我們可以指定消息隊(duì)列選擇器，而這個(gè)選擇器是我們自己實(shí)現(xiàn)了MessageQueueSelector接口定義的。

在定義選擇器的選擇算法時(shí)，一般需要使用選擇key。這個(gè)選擇key可以是消息key也可以是其它數(shù)據(jù)。但無論誰做選擇key，都不能重復(fù)，都是唯一的。

一般性的選擇算法是，讓選擇key（或其hash值）與該Topic所包含的Queue的數(shù)量取模，其結(jié)果即為選擇出的Queue的QueueId。

取模算法存在一個(gè)問題：不同選擇key與Queue數(shù)量取模結(jié)果可能會(huì)是相同的，即不同選擇key的消息可能會(huì)出現(xiàn)在相同的Queue，即同一個(gè)Consuemr可能會(huì)消費(fèi)到不同選擇key的消息。這個(gè)問題如何解決？一般性的作法是，從消息中獲取到選擇key，對(duì)其進(jìn)行判斷。若是當(dāng)前Consumer需要消費(fèi)的消息，則直接消費(fèi)，否則，什么也不做。這種做法要求選擇key要能夠隨著消息一起被Consumer獲取到。此時(shí)使用消息key作為選擇key是比較好的做法。

以上做法會(huì)不會(huì)出現(xiàn)如下新的問題呢？不屬于那個(gè)Consumer的消息被拉取走了，那么應(yīng)該消費(fèi)該消息的Consumer是否還能再消費(fèi)該消息呢？同一個(gè)Queue中的消息不可能被同一個(gè)Group中的 不同Consumer同時(shí)消費(fèi)。所以，消費(fèi)現(xiàn)一個(gè)Queue的不同選擇key的消息的Consumer一定屬于不同的Group。而不同的Group中的Consumer間的消費(fèi)是相互隔離的，互不影響的。

三、延時(shí)消息

1 什么是延時(shí)消息

當(dāng)消息寫入到Broker后，在指定的時(shí)長后才可被消費(fèi)處理的消息，稱為延時(shí)消息。

采用RocketMQ的延時(shí)消息可以實(shí)現(xiàn)定時(shí)任務(wù)的功能，而無需使用定時(shí)器。典型的應(yīng)用場景是，電商交易中超時(shí)未支付關(guān)閉訂單的場景，12306平臺(tái)訂票超時(shí)未支付取消訂票的場景。

在電商平臺(tái)中，訂單創(chuàng)建時(shí)會(huì)發(fā)送一條延遲消息。這條消息將會(huì)在30分鐘后投遞給后臺(tái)業(yè)務(wù)系 統(tǒng)（Consumer），后臺(tái)業(yè)務(wù)系統(tǒng)收到該消息后會(huì)判斷對(duì)應(yīng)的訂單是否已經(jīng)完成支付。如果未完 成，則取消訂單，將商品再次放回到庫存；如果完成支付，則忽略。

在12306平臺(tái)中，車票預(yù)訂成功后就會(huì)發(fā)送一條延遲消息。這條消息將會(huì)在45分鐘后投遞給后臺(tái)業(yè)務(wù)系統(tǒng)（Consumer），后臺(tái)業(yè)務(wù)系統(tǒng)收到該消息后會(huì)判斷對(duì)應(yīng)的訂單是否已經(jīng)完成支付。如 果未完成，則取消預(yù)訂，將車票再次放回到票池；如果完成支付，則忽略。

2 延時(shí)等級(jí)

延時(shí)消息的延遲時(shí)長不支持隨意時(shí)長的延遲，是通過特定的延遲等級(jí)來指定的。延時(shí)等級(jí)定義在RocketMQ服務(wù)端的MessageStoreConfig類中的如下變量中：

即，若指定的延時(shí)等級(jí)為3，則表示延遲時(shí)長為10s，即延遲等級(jí)是從1開始計(jì)數(shù)的。

當(dāng)然，如果需要自定義的延時(shí)等級(jí)，可以通過在broker加載的配置中新增如下配置（例如下面增加了1天這個(gè)等級(jí)1d）。配置文件在RocketMQ安裝目錄下的conf目錄中。

messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d

3 延時(shí)消息實(shí)現(xiàn)原理

具體實(shí)現(xiàn)方案是：

修改消息

Producer將消息發(fā)送到Broker后，Broker會(huì)首先將消息寫入到commitlog文件，然后需要將其分發(fā)到相應(yīng)的consumequeue。不過，在分發(fā)之前，系統(tǒng)會(huì)先判斷消息中是否帶有延時(shí)等級(jí)。若沒有，則直接正

常分發(fā)；若有則需要經(jīng)歷一個(gè)復(fù)雜的過程：

• 修改消息的Topic為SCHEDULE_TOPIC_XXXX
• 根據(jù)延時(shí)等級(jí)，在consumequeue目錄中SCHEDULE_TOPIC_XXXX主題下創(chuàng)建出相應(yīng)的queueId目錄與consumequeue文件（如果沒有這些目錄與文件的話）。

延遲等級(jí)delayLevel與queueId的對(duì)應(yīng)關(guān)系為queueId = delayLevel -1 需要注意，在創(chuàng)建queueId目錄時(shí)，并不是一次性地將所有延遲等級(jí)對(duì)應(yīng)的目錄全部創(chuàng)建完畢，而是用到哪個(gè)延遲等級(jí)創(chuàng)建哪個(gè)目錄

• 修改消息索引單元內(nèi)容。索引單元中的Message Tag HashCode部分原本存放的是消息的Tag的Hash值。現(xiàn)修改為消息的投遞時(shí)間。投遞時(shí)間是指該消息被重新修改為原Topic后再次被寫入到commitlog中的時(shí)間。投遞時(shí)間 = 消息存儲(chǔ)時(shí)間 + 延時(shí)等級(jí)時(shí)間。消息存儲(chǔ)時(shí)間指的是消息被發(fā)送到Broker時(shí)的時(shí)間戳。
• 將消息索引寫入到SCHEDULE_TOPIC_XXXX主題下相應(yīng)的consumequeue中

SCHEDULE_TOPIC_XXXX目錄中各個(gè)延時(shí)等級(jí)Queue中的消息是如何排序的？

是按照消息投遞時(shí)間排序的。一個(gè)Broker中同一等級(jí)的所有延時(shí)消息會(huì)被寫入到consumequeue目錄中SCHEDULE_TOPIC_XXXX目錄下相同Queue中。即一個(gè)Queue中消息投遞時(shí)間的延遲等級(jí)時(shí)間是相同的。那么投遞時(shí)間就取決于于消息存儲(chǔ)時(shí)間了。即按照消息被發(fā)送到Broker的時(shí)間進(jìn)行排序的。

投遞延時(shí)消息

Broker內(nèi)部有?個(gè)延遲消息服務(wù)類ScheuleMessageService，其會(huì)消費(fèi)SCHEDULE_TOPIC_XXXX中的消息，即按照每條消息的投遞時(shí)間，將延時(shí)消息投遞到?標(biāo)Topic中。不過，在投遞之前會(huì)從commitlog

中將原來寫入的消息再次讀出，并將其原來的延時(shí)等級(jí)設(shè)置為0，即原消息變?yōu)榱艘粭l不延遲的普通消息。然后再次將消息投遞到目標(biāo)Topic中。

ScheuleMessageService在Broker啟動(dòng)時(shí)，會(huì)創(chuàng)建并啟動(dòng)一個(gè)定時(shí)器TImer，用于執(zhí)行相應(yīng)的定時(shí)任務(wù)。系統(tǒng)會(huì)根據(jù)延時(shí)等級(jí)的個(gè)數(shù)，定義相應(yīng)數(shù)量的TimerTask，每個(gè)TimerTask負(fù)責(zé)一個(gè)延遲 等級(jí)消息的消費(fèi)與投遞。每個(gè)TimerTask都會(huì)檢測相應(yīng)Queue隊(duì)列的第一條消息是否到期。若第 一條消息未到期，則后面的所有消息更不會(huì)到期（消息是按照投遞時(shí)間排序的）；若第一條消息到期了，則將該消息投遞到目標(biāo)Topic，即消費(fèi)該消息。

將消息重新寫入commitlog

延遲消息服務(wù)類ScheuleMessageService將延遲消息再次發(fā)送給了commitlog，并再次形成新的消息索引條目，分發(fā)到相應(yīng)Queue。

這其實(shí)就是一次普通消息發(fā)送。只不過這次的消息Producer是延遲消息服務(wù)類 ScheuleMessageService。

四、事務(wù)消息

1 問題引入

這里的一個(gè)需求場景是：工行用戶A向建行用戶B轉(zhuǎn)賬1萬元。

我們可以使用同步消息來處理該需求場景：

• 工行系統(tǒng)發(fā)送一個(gè)給B增款1萬元的同步消息M給Broker
• 消息被Broker成功接收后，向工行系統(tǒng)發(fā)送成功ACK
• 工行系統(tǒng)收到成功ACK后從用戶A中扣款1萬元
• 建行系統(tǒng)從Broker中獲取到消息M
• 建行系統(tǒng)消費(fèi)消息M，即向用戶B中增加1萬元

這其中是有問題的：若第3步中的扣款操作失敗，但消息已經(jīng)成功發(fā)送到了Broker。對(duì)于MQ來 說，只要消息寫入成功，那么這個(gè)消息就可以被消費(fèi)。此時(shí)建行系統(tǒng)中用戶B增加了1萬元。出 現(xiàn)了數(shù)據(jù)不一致問題。

2 解決思路

解決思路是，讓第1、2、3步具有原子性，要么全部成功，要么全部失敗。即消息發(fā)送成功后，必須要保證扣款成功。如果扣款失敗，則回滾發(fā)送成功的消息。而該思路即使用事務(wù)消息。這里要使用分布式事務(wù)解決方案。

使用事務(wù)消息來處理該需求場景：

• 事務(wù)管理器TM向事務(wù)協(xié)調(diào)器TC發(fā)起指令，開啟全局事務(wù)
• 工行系統(tǒng)發(fā)一個(gè)給B增款1萬元的事務(wù)消息M給TC
• TC會(huì)向Broker發(fā)送半事務(wù)消息prepareHalf，將消息M預(yù)提交到Broker。此時(shí)的建行系統(tǒng)是看不到Broker中的消息M的
• Broker會(huì)將預(yù)提交執(zhí)行結(jié)果Report給TC。
• 如果預(yù)提交失敗，則TC會(huì)向TM上報(bào)預(yù)提交失敗的響應(yīng)，全局事務(wù)結(jié)束；如果預(yù)提交成功，TC會(huì)調(diào)用工行系統(tǒng)的回調(diào)操作，去完成工行用戶A的預(yù)扣款1萬元的操作
• 工行系統(tǒng)會(huì)向TC發(fā)送預(yù)扣款執(zhí)行結(jié)果，即本地事務(wù)的執(zhí)行狀態(tài)
• TC收到預(yù)扣款執(zhí)行結(jié)果后，會(huì)將結(jié)果上報(bào)給TM。

預(yù)扣款執(zhí)行結(jié)果存在三種可能性：

// 描述本地事務(wù)執(zhí)行狀態(tài) 
public enum LocalTransactionState { 
    COMMIT_MESSAGE, // 本地事務(wù)執(zhí)行成功 
    ROLLBACK_MESSAGE, // 本地事務(wù)執(zhí)行失敗 
    UNKNOW, // 不確定，表示需要進(jìn)行回查以確定本地事務(wù)的執(zhí)行結(jié)果 
}

• TM會(huì)根據(jù)上報(bào)結(jié)果向TC發(fā)出不同的確認(rèn)指令

若預(yù)扣款成功（本地事務(wù)狀態(tài)為COMMIT_MESSAGE），則TM向TC發(fā)送Global Commit指令

若預(yù)扣款失敗（本地事務(wù)狀態(tài)為ROLLBACK_MESSAGE），則TM向TC發(fā)送Global Rollback指令

若現(xiàn)未知狀態(tài)（本地事務(wù)狀態(tài)為UNKNOW），則會(huì)觸發(fā)工行系統(tǒng)的本地事務(wù)狀態(tài)回查操作。回查操作會(huì)將回查結(jié)果，即COMMIT_MESSAGE或ROLLBACK_MESSAGE Report給TC。TC將結(jié)果上 報(bào)給TM，TM會(huì)再向TC發(fā)送最終確認(rèn)指令Global Commit或Global Rollback

• TC在接收到指令后會(huì)向Broker與工行系統(tǒng)發(fā)出確認(rèn)指令

TC接收的若是Global Commit指令，則向Broker與工行系統(tǒng)發(fā)送Branch Commit指令。此時(shí) Broker中的消息M才可被建行系統(tǒng)看到；此時(shí)的工行用戶A中的扣款操作才真正被確認(rèn) TC接收到的若是Global Rollback指令，則向Broker與工行系統(tǒng)發(fā)送Branch Rollback指令。此時(shí) Broker中的消息M將被撤銷；工行用戶A中的扣款操作將被回滾

以上方案就是為了確保消息投遞與扣款操作能夠在一個(gè)事務(wù)中，要成功都成功，有一個(gè)失敗，則全部回滾。

以上方案并不是一個(gè)典型的XA模式。因?yàn)?em>XA模式中的分支事務(wù)是異步的，而事務(wù)消息方案中的消息預(yù)提交與預(yù)扣款操作間是同步的。

3 基礎(chǔ)

分布式事務(wù)

對(duì)于分布式事務(wù)，通俗地說就是，一次操作由若干分支操作組成，這些分支操作分屬不同應(yīng)用，分布在 不同服務(wù)器上。分布式事務(wù)需要保證這些分支操作要么全部成功，要么全部失敗。分布式事務(wù)與普通事務(wù)一樣，就是為了保證操作結(jié)果的一致性。

事務(wù)消息

RocketMQ提供了類似X/Open XA的分布式事務(wù)功能，通過事務(wù)消息能達(dá)到分布式事務(wù)的最終一致。XA 是一種分布式事務(wù)解決方案，一種分布式事務(wù)處理模式。

半事務(wù)消息

暫不能投遞的消息，發(fā)送方已經(jīng)成功地將消息發(fā)送到了Broker，但是Broker未收到最終確認(rèn)指令，此時(shí) 該消息被標(biāo)記成“暫不能投遞”狀態(tài)，即不能被消費(fèi)者看到。處于該種狀態(tài)下的消息即半事務(wù)消息。

本地事務(wù)狀態(tài)

Producer回調(diào)操作執(zhí)行的結(jié)果為本地事務(wù)狀態(tài)，其會(huì)發(fā)送給TC，而TC會(huì)再發(fā)送給TM。TM會(huì)根據(jù)TC發(fā)送來的本地事務(wù)狀態(tài)來決定全局事務(wù)確認(rèn)指令。

消息回查

消息回查，即重新查詢本地事務(wù)的執(zhí)行狀態(tài)。本例就是重新到DB中查看預(yù)扣款操作是否執(zhí)行成功。

注意，消息回查不是重新執(zhí)行回調(diào)操作。回調(diào)操作是進(jìn)行預(yù)扣款操作，而消息回查則是查看預(yù) 扣款操作執(zhí)行的結(jié)果。

引發(fā)消息回查的原因最常見的有兩個(gè)：

1)回調(diào)操作返回UNKNWON

2)TC沒有接收到TM的最終全局事務(wù)確認(rèn)指令

RocketMQ中的消息回查設(shè)置

關(guān)于消息回查，有三個(gè)常見的屬性設(shè)置。它們都在broker加載的配置文件中設(shè)置，例如：

• transactinotallow=20，指定TM在20秒內(nèi)應(yīng)將最終確認(rèn)狀態(tài)發(fā)送給TC，否則引發(fā)消息回查。默認(rèn)為60秒
• transactinotallow=5，指定最多回查5次，超過后將丟棄消息并記錄錯(cuò)誤日志。默認(rèn)15次。
• transactinotallow=10，指定設(shè)置的多次消息回查的時(shí)間間隔為10秒。默認(rèn)為60秒。

4 XA模式三劍客

XA協(xié)議

XA（Unix Transaction）是一種分布式事務(wù)解決方案，一種分布式事務(wù)處理模式，是基于XA協(xié)議的。

XA協(xié)議由Tuxedo（Transaction for Unix has been Extended for Distributed Operation，分布式操作擴(kuò)展之后的Unix事務(wù)系統(tǒng)）首先提出的，并交給X/Open組織，作為資源管理器與事務(wù)管理器的接口標(biāo)

準(zhǔn)。

XA模式中有三個(gè)重要組件：TC、TM、RM。

TC

Transaction Coordinator，事務(wù)協(xié)調(diào)者。維護(hù)全局和分支事務(wù)的狀態(tài)，驅(qū)動(dòng)全局事務(wù)提交或回滾。

RocketMQ中Broker充當(dāng)著TC。

TM

Transaction Manager，事務(wù)管理器。定義全局事務(wù)的范圍：開始全局事務(wù)、提交或回滾全局事務(wù)。它 實(shí)際是全局事務(wù)的發(fā)起者。

RocketMQ中事務(wù)消息的Producer充當(dāng)著TM。

RM

Resource Manager，資源管理器。管理分支事務(wù)處理的資源，與TC交談以注冊分支事務(wù)和報(bào)告分支事務(wù)的狀態(tài)，并驅(qū)動(dòng)分支事務(wù)提交或回滾。

RocketMQ中事務(wù)消息的Producer及Broker均是RM。

5 注意

事務(wù)消息不支持延時(shí)消息

對(duì)于事務(wù)消息要做好冪等性檢查，因?yàn)槭聞?wù)消息可能不止一次被消費(fèi)（因?yàn)榇嬖诨貪L后再提交的情況）

五、批量消息

1 批量發(fā)送消息

發(fā)送限制

生產(chǎn)者進(jìn)行消息發(fā)送時(shí)可以一次發(fā)送多條消息，這可以大大提升Producer的發(fā)送效率。不過需要注意以 下幾點(diǎn)：

• 批量發(fā)送的消息必須具有相同的Topic
• 批量發(fā)送的消息必須具有相同的刷盤策略
• 批量發(fā)送的消息不能是延時(shí)消息與事務(wù)消息

批量發(fā)送大小

默認(rèn)情況下，一批發(fā)送的消息總大小不能超過4MB字節(jié)。如果想超出該值，有兩種解決方案：

方案一：將批量消息進(jìn)行拆分，拆分為若干不大于4M的消息集合分多次批量發(fā)送

方案二：在Producer端與Broker端修改屬性

** Producer端需要在發(fā)送之前設(shè)置Producer的maxMessageSize屬性

** Broker端需要修改其加載的配置文件中的maxMessageSize屬性

生產(chǎn)者發(fā)送的消息大小

生產(chǎn)者通過send()方法發(fā)送的Message，并不是直接將Message序列化后發(fā)送到網(wǎng)絡(luò)上的，而是通過這 個(gè)Message生成了一個(gè)字符串發(fā)送出去的。這個(gè)字符串由四部分構(gòu)成：Topic、消息Body、消息日志（占20字節(jié)），及用于描述消息的一堆屬性key-value。這些屬性中包含例如生產(chǎn)者地址、生產(chǎn)時(shí)間、要發(fā)送的QueueId等。最終寫入到Broker中消息單元中的數(shù)據(jù)都是來自于這些屬性。

2 批量消費(fèi)消息

修改批量屬性

Consumer的MessageListenerConcurrently監(jiān)聽接口的consumeMessage()方法的第一個(gè)參數(shù)為消息列 表，但默認(rèn)情況下每次只能消費(fèi)一條消息。若要使其一次可以消費(fèi)多條消息，則可以通過改Consumer的consumeMessageBatchMaxSize屬性來指定。不過，該值不能超過32。因?yàn)槟J(rèn)情況下消 費(fèi)者每次可以拉取的消息最多是32條。若要修改一次拉取的最大值，則可通過修改Consumer的pullBatchSize屬性來指定。

存在的問題

Consumer的pullBatchSize屬性與consumeMessageBatchMaxSize屬性是否設(shè)置的越大越好？當(dāng)然不是。 pullBatchSize值設(shè)置的越大，Consumer每拉取一次需要的時(shí)間就會(huì)越長，且在網(wǎng)絡(luò)上傳輸出現(xiàn) 問題的可能性就越高。若在拉取過程中若出現(xiàn)了問題，那么本批次所有消息都需要全部重新拉 取。

，你consumeMessageBatchMaxSize值設(shè)置的越大，Consumer的消息并發(fā)消費(fèi)能力越低，且這批被消費(fèi)的消息具有相同的消費(fèi)結(jié)果。因?yàn)閏onsumeMessageBatchMaxSize指定的一批消息只會(huì)使用一 個(gè)線程進(jìn)行處理，且在處理過程中只要有一個(gè)消息處理異常，則這批消息需要全部重新再次消費(fèi) 處理。