團隊在日常工作中，一般情況下使用的消息隊列是騰訊雲 CKafka。CKafka 提供了高可靠的開箱即用消息隊列能力，讓我們在日常能夠放心使用，減少花在運維上的投入。不過即便如此，我們還是需要學習 Kafka 的一些基本概念和功能，從而在實際應用中嗯能夠充分高效、高質量地利用 Kafka 的能力。業務基本概念本小節主要説明的是在軟件業務層面，我們使用 Kafka 中會接觸到的概念消息 Message　　對於一個消息隊列系統，最基礎的自然是 “消息”。在 Kafka 中，“消息” 就是 Message，是在 Kafka 中數據傳輸的基本單位。多個消息會被分批次寫入 Kafka 中。這同一批次的消息，就稱為一組消息。生產者和消費者　　生產者和消費者的概念就很好理解了：產生消息的服務就稱為 “生產者” Producer，也稱為 “發佈者” Publisher 或 Writer。　　而需要獲取消息的服務就稱為 “消費者” Consumer，也稱為 “訂閲者” SubScriber 或 Reader。主題 Topic 和 分區 Partition　　在 Kafka 中，所有的消息並不是還有一條隊列。Kafka 的消息通過 Topic 進行分類。而一個 Topic 可以被區分為多個 “分區” Partitions。　　這裏需要注意的是，在 同一個 partition 內部，消息的順序是能夠保證的。也就是説：如果消息A到達 partition 的時間早於消息B，那麼消費者在獲取消息的時候，必然是先獲得消息A之後，才能夠獲取到消息B。　　但是，如果在多個 partitions 之間，消息的順序就無法保證了。比如當消費者監聽多個 partitions 時的話，消息A和消息B被讀取出來的時間無法保證。　　那這麼一來，partition 有什麼用呢？實際上 Partition 是用來做負載均衡的。當 comsumer 將消息發到一個 topic 上時，Kafka 默認會將消息儘量均衡地分發到多個 partitions 上。作為消費者監聽 topic 時，需要配置監聽哪些 partitions。一個 consumer 可以監聽多個 partitions，comsumer 和 partition(s) 的對應關係也稱為 “所有權關係”。偏移量 Offset　　Offset 是一個遞增的整數值，由 Kafka 自動遞增後自動寫入每一個 partition 中。在同一個 partition 中，一個 offset 值唯一對應着一條 message。此外，由於 offset 是遞增的，因此也可以用來區分多個 message 之間的順序。Consumer 的重啓動作並不影響 offset 的值，因為這是 Kafka 來進行維護的數值。Broker 和 集羣　　一個獨立的 Kafka server 就稱為一個 broker。一個或多個 broker 可以組成一個 “集羣” broker cluster。Kafka 雖然是一個分佈式的消息隊列系統，但是在集羣中，Kafka 依然是準中心化的系統架構。也就是説每一個集羣中依然是有一台主 broker，稱為 controller。每一個 cluster 會自動選舉一個 cluster controller 出來，controller 需要負責以下操作：管理 cluster將 partition 分配給 broker 和監控 broker。　　在 cluster 中，一個 partition 會從屬於一個 broker，這個 broker 也會稱作該 partition 的 leader。同時該 partition 也可以分配給多個 broker，進行 分區複製——如果其中一個 broker 失效了，那麼其餘的 broker 可以儘快接管 leader 的位置。如果是使用雲原生的 Kafka，我們一般就不需要太擔心這個問題。安裝/運維基本概念Kafka 部署架構　　如果是運維自己安裝 Kafka 的話，需要提前安裝的軟件是 Java 和 Zookeeper。我當時就非常疑惑怎麼多了一個 Zookeeper？實際上 Kafka 是使用 Zookeeper 來保存 cluster 中的元數據和消費者信息。這裏體現出了 Java 強大和完善的生態圈，各種方案都能夠找到已有的輪子。

​　　Zookeeper 也支持集羣部署。Zookeeper 集羣稱為 “羣組” Ensemble。因為 ensemble 也是使用了選舉機制，因此每個 ensemble 中有奇數個節點，不建議超過7個。如果我們使用了雲原生的 Kafka，就不需要過多關心這個細節啦。Topic 參數　　部署好了 broker 和 Zookeeper 之後，我們就可以創建 topics 了。創建 topic 時有一些參數需要進行配置。主要的有以下幾項需要特別留意：num.partitions: 新 topic 默認的分區數。在後續運維中，partition 的數量只會增加，不會減少。在騰訊雲 CKafka 中，這對應着 “分區數” 配置log.retension.ms: 按照時間決定 topic 中的數據可以保存多久。這對應 CKafka 界面中的 “retension.ms” 參數log.retension.bytes: 按照存儲空間決定 topic 中的數據可以保存多少。該參數在 CKafka 中不支持log.segment.bytes: 表示按照存儲空間決定日誌片段文件的大小。該參數在 CKafka 中不支持log.segment.ms: 表示按照時長決定日誌片段大小。對應 CKafka 界面中的 “segment.ms” 參數。不是必要參數​

如何選擇 Partitions 的數量？　　前面提到，在同一個 partition 中，消息的順序是能夠得到保證的。因此對於一個小型的、對可靠性要求不高、但是對順序性要求很高的系統而言，或許可以使用單 partition 的方案。　　但是這個方案其實是非常危險的：首先，單一 partition 就意味着 consumer 也只能有一個，否則會出現消息重複消費的問題。在一個生產項目中進行單點部署，這幾乎是不可接受的雖然在 Kafka 內部，單一 partition 內的消息順序能夠得到保證，但如果生產者未能得到保證的話，那麼 kafka 內的消息順序依然不是真實的。因此對於有強順序要求的消息隊列系統中，不建議使用時間順序，而是採用邏輯順序/邏輯時鐘來區分消息的先後。因此在實際生產環境中，我們應當適當地分配 partition 的數量。如果對順序性有要求，那麼不應該依賴 kafka 的順序機制，而是使用額外的機制來保證。Kafka 生產者架構圖　　生產者向 Kafka broker 發送消息一般是用各語言的 SDK 來完成的。下面框圖中是 SDK 完成的邏輯。首先 producer 在發送 message 之前，需要將 message 封裝到 producer record 中，record 包含的必填信息是 topic 和 value（也就是 message 正文）信息。此外還可選 partition 和 key 信息，不過相對少用。Key 參數的作用後文會作説明。​

　　當消息被寫入 Kafka broker 之後，broker 會回調到 SDK 中，將消息最終落地的 partition 和 partition 中的 offset 信息返回給 SDK，並最終視需要返回給 Producer。消息發送　　Kafka 生產者有兩種消息發送方式：同步 和 異步。　　同步發送方式就是生產者發出的每一個消息，都需要按照上面的結構圖的流程處理：消息發出後等待 Kafka broker 的結果響應之後再做進一步的處理。Kafka broker 返回的錯誤中包含了兩種錯誤：可重試錯誤: 當遇到這一類錯誤時，生產者可以直接重新嘗試發送。比如網絡錯誤、集羣錯誤等等。不可重試錯誤: 當遇到這一類錯誤時，生產者只能考慮告警、記錄、修改軟件邏輯等等。比如消息過大等等。異步發送方式就是生產者通過 SDK 發送消息之後就直接返回；SDK 在後台處理消息的發送、響應處理，然後通過回調告知生產者以進行進一步的處理。生產者參數　　生產者啓動之前也有一些參數可進行配置。讀者可以在各語言的 SDK 中具體查找：acks: 消息發送給 Kafka broker，由於實際上會有多個 broker，因此消息是需要複製多份的。該參數表示需要等待多少個 broker 的響應，才視為消息發送成功:0: 表示不需要等待 broker 響應1: 表示 leader 響應即可all: 表示需要所有的 broker 響應buffer.memory: 生產者的緩衝區大小。單位是 message 的數量。當緩衝區滿了之後，SDK 會根據 maxblock.ms 等待並阻塞一段時間之後再進行重試。如果緩衝區還是滿了的狀態，則會拋出異常或返回錯誤compression.type: 消息壓縮格式，可選值為: snappy, gzip, lz4retries: 重試次數，重試間隔為 retry.backoff.ms，默認是 100msbatch.size: 一個批次的數據大小，字節數。為了減少網絡傳輸中的消耗，Kafka 生產者並不是一個消息就通過一次發送發出去，而是組成一個個批次進行發送。當一個批次的大小達到這個參數時，則會馬上發出。linger.ms: 一個批次發送之前的緩衝時間。當批次的尺寸未達到 batch.size 的話，SDK 也不會一直按住 message 不發送，而是等待一段時間之後也會把內存中的批次發出client id: 自定義字符串，用於標識生產者max.id.flight.requests.per.connection: 這個參數指的是收到服務器響應之前，生產者可以發送的消息數。設置為 1 可以保證消息順序，但是相應的效率就下降了request.timeout.ms: 生產者發送數據之後等待響應的時間Key 的作用　　在 producer record 中的 key 有兩個用途：作為消息的附加消息可以用來決定寫入到哪一個分區。默認分區器可以使擁有相同 key 的消息寫入同一個分區。如果 key == null，則默認採用輪詢方式寫入分區如果 key 非空，則根據哈希結果決定分區生產者也可以通過自定義分區器來實現業務的具體分區功能，具體參見各語言的 SDKKafka 消費者　　一個 Kafka 的消費者是從屬於其對應的 comsumer group 的，每一個 group 訂閲一個 topic，每個 consumer 消費一部分的消息。整個 group 內部通過消費不同的 partition 實現負載均衡。每一個 group 都有一個 group.id 用於標識一個消費者羣組，這在業務中就對應着一個消費者業務。　　不要讓消費者數量多於分區數量，否則會導致出現重複消費的問題。因此在 partition 選用時，宜多不宜省。更多的分區數量也能夠更加合理地分配 consumer 之間的負載。分區再均衡 Partitions Reoke / Rebalance　　每個消費者可以對應一個或多個 partition；多個 consumer 組成一個 group，覆蓋 topic 的全部 partitions。但是當 consumer 和 partitions 數量發生變化時，需要重新分配所有權關係。這個動作就稱為 Rebalance。至於是熱切換還是冷切換，則由業務方決定。　　消費者在調用 subscribe() 監聽消息時，可以傳入一個 ConsumerRebalanceListener 實例來監聽事件。其中需要關注的事件有：onPartitionsRevoked(): 這是再平衡開始之前的事件。注意此時消費者應停止消費，並且 commit 已完成但尚未 commit 的 offset 值onPartitionsAssigned(): 這是再平衡結束，也就是重新分配分區結束之後的時間。大部分情況下消費者也不需要特別處理什麼，不過可以在這裏進行一些消費過程的重啓動作Commit 和 Offset　　前文提到，一個 message 能夠與 kafka 中的一個 partition 中的一個 offset 值一一對應。對於消費者而言，partition-offset 對也可以用於標識當前 comsumer 已經獲取到的消息的進度，也可以用於消費者在 kafka 中進行歷史消息的尋址。　　當對某個 message 消費完成後，消費者會將 offset 值提交到 kafka 中，從而讓 Kafka 識別並保存某個 comsumer group 的消費進度。下一次 consumer 再請求事件時，默認會從該 offset 往後繼續獲取。Consummer 向 Kafka 更新 offset 的這一動作就稱為 “提交” commit。　　如果 consumer 發生崩潰，或者有新的 consumer 加入 group，就會觸發 rebalance。完成 rebalancing 之後，每個 consumer 有可能會被分配到不同的分區。為了能夠繼續之前的工作，consumer 需要讀取每一個分區最後一次提交的 offset，然後從指定的 offset 繼續處理。這個操作，一般在 SDK 中就完成了。但是在上述切換過程中，由於分佈式系統的分佈式、異步特性，我們不可避免的還是可能遇到一些不一致的情況，具體表現為消息的重複處理和漏處理。所以我們在任何時候都不能簡單依賴 Kafka 本身提供的消息隊列機制，而是在各自的業務系統中也需要進行一定的防禦式編程，避免錯誤處理出現。　　一般而言，SDK 有下面幾種 commit 方式：自動提交: enable.auto.commit 為 true 時，API 定時、異步地進行 commit。因此，如果在觸發了再均衡的時候還有部分數據未 commit，那麼在再均衡之後在其他的消費者中就有可能發生重複消費主動提交: enable.auto.commit 為 false 時，業務方需要主動調用相關 API 進行 commit。（主動的）異步提交: 其實就是主動提交的異步版，簡單而言就是開一個後台異步 commit 的過程。提交特定的 offset: 這種模式就是顯式地 commit 具體 partition 的某個 offset 值。