消費消息基本流程

云消息隊列 Kafka 版訂閱者在訂閱消息時的基本流程為：Poll數據→執行消費邏輯→再次Poll數據，詳情參見下圖。

負載均衡

每個Group可以包含多個消費實例，即可以啟動多個云消息隊列 Kafka 版Consumer，并把參數group.id設置成相同的值。屬于同一個Group的消費實例會負載消費訂閱的Topic。

例如Group A訂閱了Topic A，并開啟三個消費實例C1、C2、C3，則發送到Topic A的每條消息最終只會傳給C1、C2、C3的某一個。云消息隊列 Kafka 版默認會均勻地把消息傳給各個消息實例，以做到消費負載均衡。

云消息隊列 Kafka 版負載均衡消費的內部原理是，把訂閱的Topic的分區，平均分配給各個消費實例。因此，消費實例的個數不要大于分區的數量，否則會有消費實例分配不到任何分區而處于空跑狀態。這個負載均衡發生的時間，除了第一次啟動上線之外，后續消費實例發生重啟、增加、減少等變更時，都會觸發一次負載均衡。

消費客戶端（Consumer）頻繁出現Rebalance

心跳超時會引發Rebalance，可以通過參數調整、提高消費速度等方法解決。更多信息，請參見為什么消費客戶端頻繁出現Rebalance？。

分區個數

分區個數主要影響的是消費者的并發數量。

對于同一個Group內的消費者來說，一個分區最多只能被一個消費者消費。因此，消費實例的個數不要大于分區的數量，否則會有消費實例分配不到任何分區而處于空跑狀態。

多個訂閱

云消息隊列 Kafka 版支持以下多個訂閱方式：

• Group訂閱多個Topic。

  一個Group可以訂閱多個Topic，多個Topic的消息被Group中的Consumer均勻消費。例如Group A訂閱了Topic A、Topic B、Topic C，則這三個Topic中的消息，被Group中的Consumer均勻消費。

  Group訂閱多個Topic的示例代碼如下：

  String topicStr = kafkaProperties.getProperty("topic");
  String[] topics = topicStr.split(",");
  for (String topic: topics) {
  subscribedTopics.add(topic.trim());
  }
  consumer.subscribe(subscribedTopics);

• Topic被多個Group訂閱。

  一個Topic可以被多個Group訂閱，且各個Group獨立消費Topic下的所有消息。例如Group A訂閱了Topic A，Group B也訂閱了Topic A，則發送到Topic A的每條消息，不僅會傳一份給Group A的消費實例，也會傳一份給Group B的消費實例，且這兩個過程相互獨立，相互沒有任何影響。

一個Group對應一個應用

建議一個Group對應一個應用，即不同的應用對應不同的代碼。如果您需要將不同的代碼寫在同一個應用中，請準備多份不同的kafka.properties。例如kafka1.properties、kafka2.properties。

消費位點

每個Topic會有多個分區，每個分區會統計當前消息的總條數，這個稱為最大位點MaxOffset。

云消息隊列 Kafka 版Consumer會按順序依次消費分區內的每條消息，記錄已經消費了的消息條數，稱為消費位點ConsumerOffset。

剩余的未消費的條數（也稱為消息堆積量）=MaxOffset-ConsumerOffset。

消費位點提交

云消息隊列 Kafka 版消費者有兩個相關參數：

• enable.auto.commit：是否采用自動提交位點機制。默認值為true，表示默認采用自動提交機制。

• auto.commit.interval.ms： 自動提交位點時間間隔。默認值為1000，即1s。

這兩個參數組合的結果就是，每次poll數據前會先檢查上次提交位點的時間，如果距離當前時間已經超過參數auto.commit.interval.ms規定的時長，則客戶端會啟動位點提交動作。

因此，如果將enable.auto.commit設置為true，則需要在每次poll數據時，確保前一次poll出來的數據已經消費完畢，否則可能導致位點跳躍。

如果想自己控制位點提交，請把enable.auto.commit設為false，并調用commit(offsets)函數自行控制位點提交。

消費位點重置

以下兩種情況，會發生消費位點重置：

• 當服務端不存在曾經提交過的位點時（例如客戶端第一次上線）。

• 當從非法位點拉取消息時（例如某個分區最大位點是10，但客戶端卻從11開始拉取消息）。

拉取大消息

消費過程是由客戶端主動去服務端拉取消息的，在拉取大消息時，需要注意控制拉取速度，注意修改配置：

• max.poll.records：每次Poll獲取的最大消息數量。如果單條消息超過1 MB，建議設置為1。

• fetch.max.bytes：設置比單條消息的大小略大一點。

• max.partition.fetch.bytes：設置比單條消息的大小略大一點。

拉取大消息的核心是逐條拉取的。

拉取公網

通過公網消費消息時，通常會因為公網帶寬的限制導致連接被斷開，此時需要注意控制拉取速度，修改配置：

1. fetch.max.bytes：建議設置成公網帶寬的一半（注意該參數的單位是bytes，公網帶寬的單位是bits）

2. max.partition.fetch.bytes：建議設置成fetch.max.bytes的三分之一或者四分之一。

消息重復和消費冪等

云消息隊列 Kafka 版消費的語義是at least once， 也就是至少投遞一次，保證消息不丟失，但是無法保證消息不重復。在出現網絡問題、客戶端重啟時均有可能造成少量重復消息，此時應用消費端如果對消息重復比較敏感（例如訂單交易類），則應該做消息冪等。

以數據庫類應用為例，常用做法是：

• 發送消息時，傳入key作為唯一流水號ID。

• 消費消息時，判斷key是否已經消費過，如果已經被消費，則忽略，如果沒消費過，則消費一次。

如果應用本身對少量消息重復不敏感，則不需要做此類冪等檢查。

消費失敗

云消息隊列 Kafka 版是按分區逐條消息順序向前推進消費的，如果消費端拿到某條消息后執行消費邏輯失敗，例如應用服務器出現了臟數據，導致某條消息處理失敗，等待人工干預，那么有以下兩種處理方式：

• 失敗后一直嘗試再次執行消費邏輯。這種方式有可能造成消費線程阻塞在當前消息，無法向前推進，造成消息堆積。

• 云消息隊列 Kafka 版沒有處理失敗消息的設計，實踐中通常會打印失敗的消息或者存儲到某個服務（例如創建一個Topic專門用來放失敗的消息），然后定時檢查失敗消息的情況，分析失敗原因，根據情況處理。

消費延遲

云消息隊列 Kafka 版的消費機制是由客戶端主動去服務端拉取消息進行消費的。因此，如果客戶端能夠及時消費，則不會產生較大延遲。如果產生了較大延遲，請先關注是否有堆積，并注意提高消費速度。

消費阻塞以及堆積

消費端最常見的問題就是消費堆積，最常造成堆積的原因是：

• 消費速度跟不上生產速度，此時應該提高消費速度，詳情請參見提高消費速度。

• 消費端產生了阻塞。

消費端拿到消息后，執行消費邏輯，通常會執行一些遠程調用，如果這個時候同步等待結果，則有可能造成一直等待，消費進程無法向前推進。

消費端應該竭力避免堵塞消費線程，如果存在等待調用結果的情況，建議設置等待的超時時間，超時后作為消費失敗進行處理。

提高消費速度

提高消費速度有以下兩個辦法：

• 增加Consumer實例個數。

  可以在進程內直接增加（需要保證每個實例對應一個線程，否則沒有太大意義），也可以部署多個消費實例進程；需要注意的是，實例個數超過分區數量后就不再能提高速度，將會有消費實例不工作。

• 增加消費線程。

  增加Consumer實例本質上也是增加線程的方式來提升速度，因此更加重要的性能提升方式是增加消費線程，最基本的步驟如下：

  1. 定義一個線程池。

  2. Poll數據。

  3. 把數據提交到線程池進行并發處理。

  4. 等并發結果返回成功后，再次poll數據執行。

消息過濾

云消息隊列 Kafka 版自身沒有消息過濾的語義。實踐中可以采取以下兩個辦法：

• 如果過濾的種類不多，可以采取多個Topic的方式達到過濾的目的。

• 如果過濾的種類多，則最好在客戶端業務層面自行過濾。

實踐中請根據業務具體情況進行選擇，也可以綜合運用上面兩種辦法。

消息廣播

云消息隊列 Kafka 版沒有消息廣播的語義，可以通過創建不同的Group來模擬實現。

訂閱關系

同一個Group內，各個消費實例訂閱的Topic最好保持一致，避免給排查問題帶來干擾。