建站
咨詢
售后
建站咨詢
新聞中心
Topic
Topic是一類消息的集合，是一種邏輯上的分區(qū)。為什么說是邏輯分區(qū)呢?因為最終數(shù)據(jù)是存儲到Broker上的，而且為了滿足高可用，采用了分布式的存儲。
這和Kafka中的實現(xiàn)如出一轍，Kafka的Topic也是一種邏輯概念，每個Topic的數(shù)據(jù)會分成很多份，然后存儲在不同的Broker上，這個「份」叫Partition。而在RocketMQ中，Topic的數(shù)據(jù)也會分布式的存儲，這個「份」叫MessageQueue。
其分布可以用下圖來表示。
這樣一來，如果某個Broker所在的機器意外宕機，而且剛好MessageQueue中的數(shù)據(jù)還沒有持久化到磁盤，那么該Topic下的這部分消息就會完全丟失。此時如果有備份的話，MQ就可以繼續(xù)對外提供服務(wù)。
為什么還會出現(xiàn)沒有持久化到磁盤的情況呢?現(xiàn)在的OS當(dāng)中，程序?qū)懭霐?shù)據(jù)到文件之后，并不會立馬寫入到磁盤，因為磁盤I/O是非常耗時的操作，在計算機來看是非常慢的一種操作。所以寫入文件的數(shù)據(jù)會先寫入到OS自己的緩存中去，然后擇機異步的將Buffer中的數(shù)據(jù)刷入磁盤。
通過多副本冗余的機制，使得RocketMQ具有了高可用的特性。除此之外，分布式存儲能夠應(yīng)對后期業(yè)務(wù)大量的數(shù)據(jù)存儲。如果不使用分布式進行存儲，那么隨著后期業(yè)務(wù)發(fā)展，消息量越來越大，單機是無論如何也滿足不了RocketMQ消息的存儲需求的。如果不做處理，那么一臺機器的磁盤總有被塞滿的時候，此時的系統(tǒng)就不具備可伸縮的特性，也無法滿足業(yè)務(wù)的使用要求了。
但是這里的可伸縮，和微服務(wù)中的服務(wù)可伸縮還不太一樣。因為在微服務(wù)中，各個服務(wù)是無狀態(tài)的。而Broker是有狀態(tài)的，每個Broker上存儲的數(shù)據(jù)都不太一樣，因為Producer在發(fā)送消息的時候會通過指定的算法，從Message Queue列表中選出一個MessageQueue發(fā)送消息。
如果不是很理解這個橫向擴展，那么可以把它當(dāng)成Redis的Cluster，通過一致性哈希，選擇到Redis Cluster中的具體某個節(jié)點，然后將數(shù)據(jù)寫入Redis Master中去。如果此時想要擴容很方便，只需要往Redis Cluster中新增Master節(jié)點就好了。
所以，數(shù)據(jù)分布式的存儲本質(zhì)上是一種數(shù)據(jù)分片的機制。在此基礎(chǔ)上，通過冗余多副本，達成了高可用。
Broker
Broker可以理解為我們微服務(wù)中的一個服務(wù)的某個實例，因為微服務(wù)中我們的服務(wù)一般來說都會多實例部署，而RocketMQ也同理，多實例部署可以幫助系統(tǒng)扛住更多的流量，也從某種方面提高了系統(tǒng)的健壯性。
在RocketMQ4.5之前，它使用主從架構(gòu)，每一個Master Broker都有一個自己的Slave Broker。
那RocketMQ的主從Broker是如何進行數(shù)據(jù)同步的呢?
Broker啟動的時候，會啟動一個定時任務(wù)，定期的從Master Broker同步全量的數(shù)據(jù)。
這塊可以先不用糾結(jié)，后面我們會通過源碼來驗證這個主從同步邏輯。
上面提到了Broker會部署很多個實例，那么既然多實例部署，那必然會存在一個問題，客戶端是如何得知自己是連接的哪個服務(wù)器?如何得知對應(yīng)的Broker的IP地址和端口?如果某個Broker突然掛了怎么辦?
NameServer
這就需要NameServer了，NameServer是什么?
這里先拿Spring Cloud舉例子——Spring Cloud中服務(wù)啟動的時候會將自己注冊到Eureka注冊中心上。當(dāng)服務(wù)實例啟動的時候，會從Eureka拉取全量的注冊表，并且之后定期的從Eureka增量同步，并且每隔30秒發(fā)送心跳到Eureka去續(xù)約。如果Eureka檢測到某個服務(wù)超過了90秒沒有發(fā)送心跳，那么就會該服務(wù)宕機，就會將其從注冊表中移除。
RocketMQ中，NameServer充當(dāng)?shù)囊彩穷愃频慕巧?。兩者從功能上也有一定的區(qū)別。
Broker在啟動的時候會向NameServer注冊自己，并且每隔30秒向NameServerv發(fā)送心跳。如果某個Broker超過了120秒沒有發(fā)送心跳，那么就會認(rèn)為該Broker宕機，就會將其從維護的信息中移除。這塊后面也會從源碼層面驗證。
當(dāng)然NameServer不僅僅是存儲了各個Broker的IP地址和端口，還存儲了對應(yīng)的Topic的路由數(shù)據(jù)。什么是路由數(shù)據(jù)呢?那就是某個Topic下的哪個Message Queue在哪臺Broker上。
Producer
總體流程
接下來，我們來看看Producer發(fā)送一條消息到Broker的時候會做什么事情，整體的流程如下。
檢查消息合法性
整體來看，其實是個很簡單的操作，跟我們平時寫代碼是一樣的，來請求了先校驗請求是否合法。Producer啟動這里會去校驗當(dāng)前Topic數(shù)據(jù)的合法性。
- Topic名稱中是否包含了非法字符
- Topic名稱長度是否超過了最大的長度限制，由常量TOPIC_MAX_LENGTH來決定，其默認(rèn)值為127
- 當(dāng)前消息體是否是NULL或者是空消息
- 當(dāng)前消息體是否超過了最大限制，由常量maxMessageSize決定，值為1024 * 1024 * 4，也就是4M。
都是些很常規(guī)的操作，和我們平時寫的checker都差不多。
獲取Topic的詳情
當(dāng)通過了消息的合法性校驗之后，就需要繼續(xù)往下走。此時的關(guān)注點就應(yīng)該從消息是否合法轉(zhuǎn)移到我要發(fā)消息給誰。
此時就需要通過當(dāng)前消息所屬的Topic拿到Topic的詳細(xì)數(shù)據(jù)。
獲取Topic的方法源碼在上面已經(jīng)給出來了，首先會從內(nèi)存中維護的一份Map中獲取數(shù)據(jù)。順帶一提，這里的Map是ConcurrentHashMap，是線程安全的，和Golang中的Sync.Map類似。
當(dāng)然，首次發(fā)送的話，這個Map肯定是空的，此時會調(diào)用NameServer的接口，通過Topic去獲取詳情的Topic數(shù)據(jù)，此時會在上面的方法中將其加入到Map中去，這樣一來下次再往該Topic發(fā)送消息就能夠直接從內(nèi)存中獲取。這里就是簡單的實現(xiàn)的緩存機制 。
從方法名稱來看，是通過Topic獲取路由數(shù)據(jù)。實際上該方法，通過調(diào)用NameServer提供的API，更新了兩部分?jǐn)?shù)據(jù)，分別是：
- Topic路由信息
- Topic下的Broker相關(guān)信息
而這兩部分?jǐn)?shù)據(jù)都來源于同一個結(jié)構(gòu)體TopicRouteData。其結(jié)構(gòu)如下。
通過源碼可以看到，就包含了該Topic下所有Broker下的Message Queue相關(guān)的數(shù)據(jù)、所有Broker的地址信息。
發(fā)送的具體Queue
此時我們獲取到了需要發(fā)送到的Broker詳情，包括地址和MessageQueue，那么此時問題的關(guān)注點又該從「消息發(fā)送給誰」轉(zhuǎn)移到「消息具體發(fā)送到哪兒」。
什么叫發(fā)送到哪兒?
開篇提到過一個Topic下會被分為很多個MessageQueue，「發(fā)送到哪兒」指的就是具體發(fā)送到哪一個Message Queue中去。
Message Queue選擇機制
核心的選擇邏輯
還是先給出流程圖
核心邏輯，用大白話講就是將一個隨機數(shù)和Message Queue的容量取模。這個隨機數(shù)存儲在Thread Local中，首次計算的時候，會直接隨機一個數(shù)。
此后，都直接從ThreadLocal中取出該值，并且+1返回，拿到了MessageQueue的數(shù)量和隨機數(shù)兩個關(guān)鍵的參數(shù)之后，就會執(zhí)行最終的計算邏輯。
接下來，我們來看看選擇Message Queue的方法SelectOneMessageQueue都做了什么操作吧。
可以看到，主邏輯被變量sendLatencyFaultEnable分為了兩部分。
容錯機制下的選擇邏輯
該變量表意為發(fā)送延遲故障。本質(zhì)上是一種容錯的策略，在原有的MessageQueue選擇基礎(chǔ)上，再過濾掉不可用的Broker，對之前失敗的Broker，按一定的時間做退避。
可以看到，如果調(diào)用Broker信息發(fā)生了異常，那么就會調(diào)用updateFault這個方法，來更新Broker的Aviable情況。注意這個參數(shù)isolation的值為true。接下來我們從源碼級別來驗證上面說的退避3000ms的事實。
可以看到，isolation值是true，則duration通過三元運算符計算出來結(jié)果為30000，也就是30秒。所以我們可以得出結(jié)論，如果發(fā)送消息拋出了異常，那么直接會將該Broker設(shè)置為30秒內(nèi)不可用。
而如果只是發(fā)送延遲較高，則會根據(jù)如下的map，根據(jù)延遲的具體時間，來判斷該設(shè)置多少時間的不可用。
例如，如果上次請求的latency超過550ms，就退避3000ms;超過1000，就退避60000;
正常情況下的選擇邏輯
而正常情況下，如果當(dāng)前發(fā)送故障延遲沒有啟用，則會走常規(guī)邏輯，同樣的會去for循環(huán)計算，循環(huán)中取到了MessageQueue之后會去判斷是否和上次選擇的MessageQueue屬于同一個Broker，如果是同一個Broker，則會重新選擇，直到選擇到不屬于同一個Broker的MessageQueue，或者直到循環(huán)結(jié)束。這也是為了將消息均勻的分發(fā)存儲，防止數(shù)據(jù)傾斜。
發(fā)送消息
選到了具體的Message Queue之后就會開始執(zhí)行發(fā)送消息的邏輯，就會調(diào)用底層Netty的接口給發(fā)送出去，這塊暫時沒啥可看的。
Broker的啟動流程
主從同步
在上面提到過，RocketMQ有自己的主從同步，但是有兩個不同的版本，版本的分水嶺是在4.5版本。這兩個版本區(qū)別是什么呢?
- 4.5之前：有點類似于Redis中，我們手動的將某臺機器通過命令slave of 變成另一臺Redis的Slave節(jié)點，這樣一來就變成了一個較為原始的一主一從的架構(gòu)。為什么說原始呢?因為如果此時Master節(jié)點宕機，我們需要人肉的去做故障轉(zhuǎn)移。RocketMQ的主從架構(gòu)也是這種情況。
- 4.5之后：引入了Dleger，可以實現(xiàn)一主多從，并且實現(xiàn)自動的故障轉(zhuǎn)移。這就跟Redis后續(xù)推出了Sentinel是一樣的。Dleger也是類似的作用。
下圖是Broker啟動代碼中的源碼。
可以看到判斷了是否開啟了Dleger，默認(rèn)是不開啟的。所以就會執(zhí)行其中的邏輯。
剛好我們就看到了，里面有Rocket主從同步數(shù)據(jù)的相關(guān)代碼。
如果當(dāng)前Broker節(jié)點的角色是Slave，則會啟動一個周期性的定時任務(wù)，定期(也就是10秒)去Master Broker同步全量的數(shù)據(jù)。同步的數(shù)據(jù)包括：
- Topic的相關(guān)配置
- Cosumer的消費偏移量
- 延遲消息的Offset
- 訂閱組的相關(guān)數(shù)據(jù)和配置
注冊Broker
完成了主動同步定時任務(wù)的啟動之后，就會去調(diào)用registerBrokerAll去注冊Broker?？赡苓@里會有點疑問，我這里是Broker啟動，只有當(dāng)前一個Broker實例，那這個All是什么意思呢?
All是指所有的NameServer，Broker啟動的時候會將自己注冊到每一個NameServer上去。為什么不只注冊到一個NameServer就完事了呢?這樣一來還可以提高效率。歸根結(jié)底還是高可用的問題。
如果Broker只注冊到了一臺NameServer上，萬一這臺NameServer掛了呢?這個Broker對所有客戶端就都不可見了。實際上Broker還在正常的運行。
進到registerBrokerAll中去。
可以看到，這里會判斷是否需要進行注冊。通過上面的截圖可以看到，此時forceRegister的值為true，而是否要注冊，決定權(quán)就交給了needRegister
為什么需要判斷是否需要注冊呢?因為Broker一旦注冊到了NameServer之后，由于Producer不停的在寫入數(shù)據(jù)，Consumer也在不停的消費數(shù)據(jù)，Broker也可能因為故障導(dǎo)致某些Topic下的Message Queue等關(guān)鍵的路由信息發(fā)生變動。
這樣一來，NameServer中的數(shù)據(jù)和Broker中的數(shù)據(jù)就會不一致。
如何判斷是否需要注冊
大致的思路是，Broker會從每一個NameServer中獲取到當(dāng)前Broker的數(shù)據(jù)，并和當(dāng)前Broker節(jié)點中的數(shù)據(jù)做對比。但凡有一臺NameServer數(shù)據(jù)和當(dāng)前Broker不一致，都會進行注冊操作。
接下來，我們從源碼層面驗證這個邏輯。關(guān)鍵的邏輯我在圖中也標(biāo)注了出來。
可以看到， 就是通過對比Broker中的數(shù)據(jù)版本和NameServer中的數(shù)據(jù)版本來實現(xiàn)的。這個版本，注冊的時候會寫到注冊的數(shù)據(jù)中存入NameServer中。
這里由于是有多個，所以RocketMQ用線程池來實現(xiàn)了多線程操作，并且用CountDownLatch來等待所有的返回結(jié)果。經(jīng)典的用空間換時間，Golang里面也有類似的操作，那就是sync.waitGroup。
關(guān)于任何一個數(shù)據(jù)不匹配，都會進行重新注冊的事實，我們也從源碼層面來驗證一下。
可以看到，如果任何一臺NameServer的數(shù)據(jù)發(fā)生了Change，都會break，返回true。
這里的結(jié)果列表使用的是CopyOnWriteList來實現(xiàn)的。
因為這里是多線程去執(zhí)行的判斷邏輯，而正常的列表不是線程安全的。CopyOnWriteArrayList之所以是線程安全的，這歸功于COW(Copy On Write)，讀請求時共用同一個List，涉及到寫請求時，會復(fù)制出一個List，并在寫入數(shù)據(jù)的時候加入獨占鎖。比起直接對所有操作加鎖，讀寫鎖的形式分離了讀、寫請求，使其互不影響，只對寫請求加鎖，降低了加鎖的次數(shù)、減少了加鎖的消耗，提升了整體操作的并發(fā)。
執(zhí)行注冊邏輯
這塊就是構(gòu)建數(shù)據(jù)，然后多線程并發(fā)的去發(fā)送請求，用CopyOnWriteArrayList來保存結(jié)果。不過，上面我們提到過，Broker注冊的時候，會把數(shù)據(jù)版本發(fā)送到NameServer并且存儲起來，這塊我們可以看看發(fā)送到NameServer的數(shù)據(jù)結(jié)構(gòu)。
可以看到，Topic的數(shù)據(jù)分為了兩部分，一部分是核心的邏輯，另一部分是DataVersion，也就是我們剛剛一直提到的數(shù)據(jù)版本。
Broker如何存儲數(shù)據(jù)
剛剛在聊Producer最后提到的是，發(fā)送消息到Broker就完了。不知道大家有沒有想過Broker是如何存儲消息的?
Commit log
先給出流程圖
然后給出結(jié)論，Producer發(fā)送的消息是存儲在一種叫commit log的文件中的，Producer端每次寫入的消息是不等長的，當(dāng)該CommitLog文件寫入滿1G，就會新建另一個新的CommitLog，繼續(xù)寫入。此次采取的是順序?qū)懭搿?/p>
那么問題來了，Consumer來消費的時候，Broker是如何快速找到對應(yīng)的消息的呢?我們首先排除遍歷文件查找的方法， 因為RocketMQ是以高吞吐、高性能著稱的，肯定不可能采取這種對于很慢的操作。那RocketMQ是如何做的呢?
答案是ConsumerQueue
ConsumerQueue
ConsumerQueue是什么?是文件。引入的目的是什么呢?提高消費的性能。
Broker在收到一條消息的時候，寫入Commit Log的同時，還會將當(dāng)前這條消息在commit log中的offset、消息的size和對應(yīng)的Tag的Hash寫入到consumer queue文件中去。
每個MessageQueue都會有對應(yīng)的ConsumerQueue文件存儲在磁盤上，每個ConsumerQueue文件包含了30W條消息，每條消息的size大小為20字節(jié)，包含了8字節(jié)CommitLog的Offset、4字節(jié)的消息長度、8字節(jié)的Tag的哈希值。這樣一來，每個ConsumerQueue的文件大小就約為5.72M。
當(dāng)該ConsumerQueue文件寫滿了之后，就會再新建一個ConsumerQueue文件，繼續(xù)寫入。
所以，ConsumerQueue文件可以看成是CommitLog文件的索引。
負(fù)載均衡
什么意思呢?假設(shè)我們總共有6個MessageQueue，然后此時分布在了3臺Broker上，每個Broker上包含了兩個queue。此時Consumer有3臺，我們可以大致的認(rèn)為每個Consumer負(fù)責(zé)2個MessageQueue的消費。但是這里有一個原則，那就是一個MessageQueue只能被一臺Consumer消費，而一臺Consumer可以消費多個MessageQueue。
為什么?道理很簡單，RocketMQ支持的順序消費，是指的分區(qū)順序性，也就是在單個MessageQueue中，消息是具有順序性的，而如果多臺Consumer去消費同一個MessageQueue，就很難去保證順序消費了。
由于有很多個Consumer在消費多個MessageQueue，所以為了不出現(xiàn)數(shù)據(jù)傾斜，也為了資源的合理分配利用，在Producer發(fā)送消息的時候，需要盡可能的將消息均勻的分發(fā)給多個MessageQueue。
同時，上面那種一個Consumer消費了2個MessageQueue的情況，萬一這臺Consumer掛了呢?這兩個MessageQueue不就沒人消費了?
以上兩種情況分別是Producer端的負(fù)載均衡、Consumer端的負(fù)載均衡。
Producer端負(fù)載均衡
關(guān)于Producer端上面的負(fù)載均衡，上面的流程圖已經(jīng)給了出來，并且給出了源碼的驗證。首先是容錯策略，會去避開一段時間有問題的Broker，并且加上如果選擇了上次的Broker，就會重新進行選擇。
Consumer端負(fù)載均衡
首先Consumer端的負(fù)責(zé)均衡可以由兩個對象觸發(fā)：
- Broker
- Consumer自身
Consumer也會向所有的Broker發(fā)送心跳，將消息的消費組名稱、訂閱關(guān)系集合、消息的通信模式和客戶端的ID等等。Broker收到了Consumer的心跳之后，會將其存在Broker維護的一個Manager中，名字叫ConsumerManager。當(dāng)Broker監(jiān)聽到了Consumer數(shù)量發(fā)生了變動，就會通知Consumer進行Rebalance。
但是如果Broker通知Consumer進行Rebalance的消息丟了呢?這也就是為什么需要對Consumer自身進行觸發(fā)的原因。Consumer會在啟動的時候啟動定時任務(wù)，周期性的執(zhí)行rebalance操作。
默認(rèn)是20秒執(zhí)行一次。具體的代碼如下。
具體流程
首先，Consumer的Rebalance會獲取到本地緩存的Topic的全部數(shù)據(jù)，然后向Broker發(fā)起請求，拉取該Topic和ConsumerGroup下的所有的消費者信息。此處的Broker數(shù)據(jù)來源就是Consumer之前的心跳發(fā)送過去的數(shù)據(jù)。然后會對Topic中MessageQueue和消費者ID進行排序，然后用消息隊列默認(rèn)分配算法來進行分配，這里的默認(rèn)分配策略是平均分配。
首先會均勻的按照類似分頁的思想，將MessageQueue分配給Consumer，如果分配的不均勻，則會依次的將剩下的MessageQueue按照排序的順序，從上往下的分配。所以在這里Consumer 1被分配到了4個MessageQueue，而Consumer 2被分配到了3個MessageQueue。
Rebalance完了之后，會將結(jié)果和Consumer緩存的數(shù)據(jù)做對比，移除不在ReBalance結(jié)果中的MessageQueue，將原本沒有的MessageQueue給新增到緩存中。
觸發(fā)時機
- Consumer啟動時 啟動之后會立馬進行Rebalance
- Consumer運行中 運行中會監(jiān)聽Broker發(fā)送過來的Rebalance消息，以及Consumer自身的定時任務(wù)觸發(fā)的Rebalance
- Consumer停止運行 停止時沒有直接地調(diào)用Rebalance，而是會通知Broker自己下線了，然后Broker會通知其余的Consumer進行Rebalance。
換一個角度來分析，其實就是兩個方面，一個是隊列信息發(fā)生了變化，另一種是消費者發(fā)生了變化。
源碼驗證
然后給出核心的代碼驗證，獲取數(shù)據(jù)的邏輯如下
驗證了我們剛剛說的獲取了本地的Topic數(shù)據(jù)緩存，和從Broker端拉取所有的ConsumerID。
接下來是驗證剛說的排序邏輯。
接下來是看判斷結(jié)果是否發(fā)生了變化的源碼。
可以看到，Consumer通知Broker策略，其本質(zhì)上就是發(fā)送心跳，將更新后的數(shù)據(jù)通過心跳發(fā)送給所有的Broker。
Consumer更多的細(xì)節(jié)
可能關(guān)于Consumer，我們使用的更多一點。例如我們知道我們可以設(shè)置集群消費和廣播消息，分別對應(yīng)RocketMQ中的CLUSTERING和BROADCASTING**。
再比如我們知道，我們可以設(shè)置順序消費和并發(fā)消費等等，接下來就讓我們用源碼來看看這些功能在RocketMQ中是怎么實現(xiàn)的。
消費模型
在Consumer中，默認(rèn)都是采用集群消費，這塊在Consumer的代碼中也有體現(xiàn)。
而消費模式的不同，會影響到管理offset的具體實現(xiàn)。
可以看到，當(dāng)消費模型是廣播模式時，Offset的持久化管理會使用實現(xiàn)LocalFileOffsetStorage
當(dāng)消費模式是集群消費時，則會使用RemoteBrokerOffsetStore。
具體原因是什么呢?首先我們得知道廣播模式和集群模式的區(qū)別在哪兒：
- 廣播模式下，一條消息會被ConsumerGroup中的每一臺機器所消費
- 集群模式下，一條消息只會被ConsumerGroup中的一臺機器消費
所以在廣播模式下，每個ConsumerGroup的消費進度都不一樣，所以需要由Consumer自身來管理Offset。而集群模式下，同個ConsumerGroup下的消費進度其實是一樣的，所以可以交由Broker統(tǒng)一管理。
消費模式
消費模式則分為順序消費和并發(fā)消費，分別對應(yīng)實現(xiàn)MessageListenerOrderly和MessageListenerConcurrently兩種方式。
不同的消費方式會采取不同的底層實現(xiàn)，配置完成之后就會調(diào)用start。
拉取消息
接下來我們來看一個跟我們最最相關(guān)的問題，那就是我們平時消費的消息到底是怎么樣從Broker發(fā)到的Consumer。在靠近啟動Rebalance的地方，Consumer也開啟了一個定時拉取消息的線程。
這個線程做了什么事呢?它會不停的從一個維護在內(nèi)存中的Queue中獲取一個在寫入的時候就構(gòu)建好的PullRequest對象，調(diào)用具體實現(xiàn)去不停的拉取消息了。
處理消費結(jié)果
在這里是否開啟AutoCommit，所做的處理差不了很多，大家也都知道，唯一區(qū)別就在于是否自動的提交Offset。對于處理成功的邏輯也差不多，我們平時業(yè)務(wù)邏輯中可能也并不關(guān)心消費成功的消息。我們更多關(guān)注的是如果消費失敗了，RocketMQ是怎么處理的?
這是在AutoCommit下，如果消費失敗了的處理邏輯。會記錄一個失敗的TPS，然后這里有一個非常關(guān)鍵的邏輯，那就是checkReconsumeTimes。
如果當(dāng)前消息的重試次數(shù)，如果大于了最大的重試消費次數(shù)，就會把消費發(fā)回給Broker。那最大重試次數(shù)是如何定義的。
如果值為-1，那么最大次數(shù)就是MAX_VALUE，也就是2147483647。這里有點奇怪啊，按照我們平常的認(rèn)知，難道不是重試16次嗎?然后就看到了很騷的一句注釋。
-1 means 16 times，這代碼確實有點，一言難盡。
然后，如果超過了最大的次數(shù)限制，就會將該消息調(diào)用Prodcuer的默認(rèn)實現(xiàn)，將其發(fā)送到死信隊列中。當(dāng)然，死信隊列也不是什么特殊的存在，就是一個單獨的Topic而已。
通過getRetryTopic來獲取的，默認(rèn)是給當(dāng)前的ConsumerGroup名稱加上一個前綴。
網(wǎng)頁名稱：RocketMQ基礎(chǔ)概念剖析&源碼解析
網(wǎng)頁鏈接：http://www.5511xx.com/article/coepigc.html
