【RocketMQ】RocketMQ 5.0新特性（二）- Pop消費模式
Pop模式消費和訊息粒度負載均衡
在RocketMQ 5.0之前，消費有兩種方式可以從Broker獲取訊息，分別為Pull模式和Push模式。
- Pull模式：消費需要不斷的從阻塞佇列中獲取資料，如果沒有資料就等待，這個阻塞佇列中的資料由訊息拉取執行緒從Broker拉取訊息之後加入的，所以Pull模式下消費需要不斷主動從Broker拉取訊息。
- Push模式：需要註冊訊息監聽器，當有訊息到達時會通過回撥函數進行訊息消費，從表面上看就像是Broker主動推播給消費者一樣，所以叫做推模式，底層依舊是消費者從Broker拉取資料然後觸發回撥函數進行訊息消費，只不過不需要像Pull模式一樣不斷判斷是否有訊息到來。
注：圖片來自RocketMQ官方檔案
不過不管是Pull模式還是Push模式，在叢集模式下，一個訊息佇列只能分配給同一個消費組內的某一個消費者進行消費，所以需要進行Rebalance負載均衡為每個消費者分配訊息佇列之後才可以進行訊息消費。
Rebalance的工作是在每個消費者端進行的，消費端負責的工作太多，除了負載均衡還有消費位點管理等功能，如果新增一種語言的支援，就需要重新實現一遍對應的業務邏輯程式碼。
除此以外，在RocketMQ 5.0以前負載均衡是以訊息佇列為維度為每個消費者分配的，一個訊息佇列只能分給組內一個消費者消費，所以會存在以下問題：
（1）佇列只能分給組內一個消費者消費，也就無法通過擴充套件消費者的數量來提升消費能力；
（2）訊息佇列數量與消費者數量比例不均衡時，可能會導致某些消費者沒有訊息佇列可以分配或者某些消費者承擔過多的訊息佇列，分配不均勻；
（3）如果某個消費者hang主，會導致分配到該消費者的訊息佇列中的訊息無法消費，導致訊息積壓；
在RocketMQ 5.0增加了Pop模式消費，將負載均衡、消費位點管理等功能放到了Broker端，減少使用者端的負擔，使其變得輕量級，並且5.0之後支援訊息粒度的負載均衡。
訊息粒度負載均衡
對於PushConsumer和SimpleConsumer型別的消費者，預設且僅使用訊息粒度負載均衡策略。
注：圖片來自RocketMQ官方檔案
訊息粒度負載均衡策略中，同一消費組內的多個消費者將按照訊息粒度平均分攤主題中的所有訊息，即同一個佇列中的訊息，可被平均分配給組內多個消費者共同消費。
訊息粒度負載均衡策略保證同一個佇列的訊息可以被組內多個消費者共同處理，但是該策略使用的訊息分配演演算法結果是隨機的，不能指定訊息被哪一個特定的消費者處理。當消費者獲取到某條訊息後，伺服器端會對該訊息加鎖，保證該訊息對其他消費者不可見，直到訊息消費成功或者超時，所以多個消費者同時消費同一個訊息佇列中的訊息，伺服器端也可以保證訊息不會被多個消費者重複消費。
訊息粒度負載均衡策略適用於絕大多數線上處理的業務場景。
Pop訊息消費
首先使用者端（消費者）向伺服器端（Broker）傳送Pop請求，Broker端收到請求後以Pop模式獲取訊息，之後返回給使用者端，使用者端消費訊息成功之後，向Broker傳送ACK請求確認訊息消費成功。
當POP出一條訊息之後，這條訊息就會在一段時間內不可見，在這個時間段內，這條訊息不會再被POP出來，如果在這個期間未能收到該訊息的ACK請求，過了這個不可見的時間之後，訊息就會恢復可見狀態，重新被消費。
POP的消費位點由Broker儲存和控制，並且POP模式可以使多個消費者端消費同一個訊息佇列中的訊息，消費者端不再需要在本地做負載均衡分配訊息佇列，只需要呼叫伺服器端提供的POP介面獲取訊息進行消費即可，即便某個消費者hang住，其他消費者依舊可以繼續消費佇列中的資料，不會造成訊息堆積。
POP訊息在Broker端的實現
-
Broker端在處理POP請求時，先在佇列維度加鎖，保證同一時間只有一個消費者可以從該佇列中獲取訊息；
-
Broker端會從佇列中獲取一批訊息，並構建這批訊息對應的CheckPoint資訊儲存在Broker中，之後會與ACK的訊息進行匹配；
CheckPoint主要包括訊息的 Topic，ConsumerGroup，QueueId，offset，POPTime，msgCout，reviveQueueId等資訊。 -
CheckPoint會優先儲存在記憶體中，如果在一段時間內收到了使用者端的ACK訊息，就會將對應的CheckPoint清除，並更新消費進度；
-
對於一段時間內為收到ACK訊息的CheckPoint，會將其從記憶體中刪除，然後傳送到延時主題
SCHEDULE_TOPIC_XXXX中，到達延時時間之後，訊息會再被轉發到REVIVE_TOPIC（會使用REVIVE_LOG_ + 叢集名稱作為主題）中，有一個執行緒去處理REVIVE_TOPIC中的資料，將裡面的訊息拉取放入到一個
MAP中，如果後續收到對應的ACK訊息，則會更新REVIVE_TOPIC主題中的消費位點標識訊息消費完成，如果過了一定時間依舊未收到對應的ACK訊息，會查詢這個CheckPoint對應的真實訊息，將其放入到重試佇列中，等待使用者端消費，所以消費者消費的時候有一定概率可以消費到重試佇列中的訊息。
由於一個訊息佇列中的訊息可以被多個消費者消費，如果某個消費者在消費某條訊息之後一直未發生ACK訊息，那麼Broker是如何管理消費進度的，比如佇列1中有1、2、3、4、5條訊息，此時有三個消費者1、2、3，分別分配到了佇列中的1、2、3條訊息，此時消費者1已經對訊息1ACK完畢，消費者3也對訊息3ACK完畢，消費者2一直未ACK訊息2，那麼Broker如何設定消費進度？
個人認為，在一段時間內訊息2對應的CheckPoint未匹配到對應的ACK訊息，為了保證消費可以繼續向後消費訊息，應該會推進消費進度跳過這個訊息，對於訊息2，會按照超時處理邏輯，將其對應的CheckPoint先放入延時佇列，再放入REVIVE_TOPIC中，之後等待ACK，如果之後一直還未收到ACK再將其放入重試佇列，等待重新消費。
參考
RocketMQ官方檔案