1. Producer

Producers 負(fù)責(zé)生產(chǎn)消息趟径。
消息成功寫(xiě)到 topic 后钮热，broker 會(huì)返回 producer 消息的 topic, partition & the offset of the record within the partition。

1.1 Send

發(fā)送方式有兩種：

• Synchronous send
• Asynchronous send

1.2 Retry

消息可能因?yàn)橐恍┊惓Ｔ驅(qū)懯∶ズ洌惓７譃閮深?lèi)：

• Retriable errors: KafkaProducer 針對(duì)這種異常庞呕，可以自動(dòng)的發(fā)起重試。全部邏輯隱藏在 send 方法中疗疟，開(kāi)發(fā)人員不需要人工干預(yù)
  • a connection error can be resolved because the connection may get reestablished.
  • A “no leader” error can be resolved when a new leader is elected for the partition.
• Nonretriable errors: 這種錯(cuò)誤沒(méi)法通過(guò)重試修復(fù)该默，會(huì)直接拋異常，需要開(kāi)發(fā)人員處理
  • message size too large error

1.3 Acks

acks 參數(shù)控制 producer 認(rèn)為 message 寫(xiě)成功之前必須接收到 partition 成功寫(xiě)入的副本數(shù)（針對(duì) replicas）策彤∷ㄐ洌可以把acks 理解為用來(lái)控制數(shù)據(jù)備份時(shí)的一致性強(qiáng)弱的。

當(dāng)配置 acks 為：

• 0: producer 不會(huì)等待 broker 的寫(xiě)成功回復(fù)店诗，producer 發(fā)完 request 直接 return裹刮，把操作權(quán)交給 application developer，這種配置可能造成消息丟失
• 1: leader replica 成功寫(xiě)消息后庞瘸，broker 響應(yīng)寫(xiě)成功通知
  如果 partition leader 所在的 broker crash 了捧弃，而新的 leader 還沒(méi)有選舉出來(lái)，則 producer 會(huì)收到 error 的 response 并發(fā)起 retry擦囊。
  如果 partition leader 寫(xiě)成功(響應(yīng) producer 成功)后 crash 了违霞，而一個(gè)還沒(méi)有同步到 message 的 replica 被選為新的 leader，那這條消息就丟失了霜第。
• all: request 將存在 buffer 中葛家，直到 leader 觀察到所有的 follower replicas 都備份完 messages，才響應(yīng) producer泌类。優(yōu)點(diǎn)：數(shù)據(jù)一致性強(qiáng)癞谒；缺點(diǎn)：性能差。

2. Brokers and Clusters

一個(gè) Kafka 的服務(wù)器叫做一個(gè) broker刃榨。broker 接受 producer 傳遞過(guò)來(lái)的 messages弹砚，store messages 到指定的 partition 中，并分配 offsets枢希。它還接受 consumer 發(fā)過(guò)來(lái)的 poll messages request & heartbeats request

broker 的 metadata 在 zookeeper 中維護(hù)桌吃。每一個(gè) broker 都配置有自身唯一的 id，當(dāng) broker start 時(shí)苞轿，broker 將自身的 id 注冊(cè)到 zk 中（通過(guò)寫(xiě)一個(gè) ephemeral node）茅诱，如果已經(jīng)存在一個(gè)相同 ID 的 ephemeral node，則會(huì)報(bào)錯(cuò)搬卒。

當(dāng) broker 和 zk 斷掉連接后（broker stop / network partition / long garbage-collection pause）瑟俭，broker 啟動(dòng)時(shí)創(chuàng)建的 ephemeral node 將自動(dòng)被 zk 刪除。

當(dāng)完全丟失 broker & 刪除對(duì)應(yīng)的 ephemeral node 后契邀，重啟一個(gè)具有相同 id 的新 broker摆寄，該 broker 將替代丟失的舊 broker，接受原 broker 相同的 partitions & topics。

2.1 The Controller

一組 brokers 可以搭建成一個(gè) cluster微饥。在 cluster 中有一個(gè) broker 擔(dān)任 cluster controller逗扒，一般是第一個(gè)加入 cluster 的 broker 擔(dān)任 controller，它會(huì)在 zk 中創(chuàng)建一個(gè)名叫 /controller 的 ephemeral node欠橘。

當(dāng) controller broker stop or loses connectivity to zk 時(shí)矩肩，它創(chuàng)建的 /controller node 會(huì)被 zk 刪除。cluster 中別的 brokers 將被通知 controller 丟失简软，剩下的 brokers 繼續(xù)搶占 /controller node蛮拔，第一個(gè)寫(xiě)成功的成為新的 controller述暂。

controller broker 除了承擔(dān)普通的 broker 功能外痹升，還負(fù)責(zé) partition leaders 的選舉。如果 controller broker 發(fā)現(xiàn)有別的 broker 離開(kāi) cluster（通過(guò)監(jiān)聽(tīng) zk 的相關(guān)路徑 node）時(shí)畦韭，那么所有存在于丟失 broker 上的 leader partitions 需要新的 leader疼蛾，controller 負(fù)責(zé)選擇一個(gè) partition 作為 leader，并通知給各個(gè) brokers partitions艺配。新的 partition leaders 明確自己的職責(zé)察郁，followers 則明確自己需要同步的 new leader。

The controller uses the epoch number to prevent a “split brain” scenario where two nodes believe each is the current controller.

2.2 Multiple Clusters

The replication mechanisms within the Kafka clusters are designed only to work within a single cluster, not between multiple clusters.
The Kafka project includes a tool called MirrorMaker, used for this purpose.

2.3 Zookeeper

kafka with zookeeper.jpg

Kafka 用 Zookeeper 來(lái)維護(hù) broker cluster转唉，存儲(chǔ) brokers, topics, partitions 的 metadata皮钠。
Consumer 的 metadata 在 kafka v0.9 之前的版本中，是通過(guò) Zookeeper 維護(hù)赠法。但是在 v0.9 之后麦轰，可以選擇通過(guò) zookeeper 管理，也可以選擇通過(guò) kafka brokers 管理砖织，因?yàn)轭l繁的讀寫(xiě) offsets 對(duì) zk 的壓力較大款侵，所以推薦通過(guò) Kafka broker 管理。

3. Replication

前段時(shí)間一直在系統(tǒng)性的學(xué)習(xí)分布式存儲(chǔ)的知識(shí)侧纯，某種程度上來(lái)說(shuō)新锈，Kafka, Redis, MySQL Cluster, Zookeeper 等都可以理解為分布式存儲(chǔ)的實(shí)現(xiàn)。而分布式存儲(chǔ)中的核心實(shí)現(xiàn)方法就是 partition & replication眶熬。前面我們介紹了 Kafka 的 topic partition妹笆，現(xiàn)在來(lái)了解下 replication。

為了實(shí)現(xiàn)系統(tǒng)的可靠性(availability) & 耐用性(durability)娜氏，我們可以對(duì) partition 做 replicated拳缠。kafka 中的 data 按 topics 組織，每個(gè) topic 可以做 partition牍白，每個(gè) partition 可以有多個(gè) replicas脊凰。

replicas 有兩種角色：

• leader replica: 每個(gè) partition 有一個(gè) leader
  • 對(duì)該 partition 的所有讀寫(xiě)操作(producer, follower replicas, consumer)都是在 leader 節(jié)點(diǎn)上完成的，這也是保持 consistency 的一種方式
  • leader 還知道每個(gè) followers 的同步進(jìn)度
• follower replica: 除 leader 外的，都是 followers狸涌，followers 不直接承擔(dān) client 的讀寫(xiě)任務(wù)切省。它們的唯一工作就是通過(guò)向 leader 發(fā) fetch request 備份 messages。

consumer.poll / replica fetch messages 時(shí)帕胆，會(huì)把自身已有的最大 offset 帶給 leader 來(lái)獲得準(zhǔn)確的 messages朝捆。當(dāng) leader crashes 時(shí)，其中一個(gè)擁有最多消息（最大 offsets）的 followers 將變成 leader懒豹。

4. Consumer

consumers 消費(fèi)消息芙盘。同時(shí) consumer 會(huì)跟蹤上報(bào)自己已消費(fèi)消息的 offset（kafka 的每個(gè)消息在 topic 的 partition 中都有一個(gè)唯一的 offset）。
consumers 是以 consumer group 的形式工作的脸秽。group 保證每個(gè) partition 只能被一個(gè) consumer 消費(fèi)儒老，換言之，group 中的 consumer 消費(fèi)互不相同的 partition记餐。
一個(gè) consumer group/ consumers 可以（通過(guò)正則表達(dá)式）訂閱多個(gè) topics驮樊，當(dāng)新增滿足正則表達(dá)式的 topic 時(shí)，能自動(dòng)讀取到該 topic 的 msg 片酝。

4.1 Consumer Groups

consumer groups 的可能組織結(jié)構(gòu)有：

consumer group.jpg

consumer group2.jpg

consumer group3.jpg

consumer group4.jpg

由于 consumer 經(jīng)常會(huì)做一些高延遲的操作囚衔，例如寫(xiě)數(shù)據(jù)庫(kù)、分析數(shù)據(jù)等雕沿，consumer 的消費(fèi)能力可能會(huì)小于 producer 的生成能力练湿。

分析以下場(chǎng)景：

• 如果 partition count > consumer count，可以往 consumer group 中加更多的 consumers 來(lái)分擔(dān)負(fù)載审轮，提升消費(fèi)能力
• 如果 partition count < consumer count肥哎，多出來(lái)的 consumers 會(huì)閑置，需要增加 partition 的數(shù)量断国。但是這帶來(lái)的問(wèn)題是：相同 key 的 messages 在增減 partition count 前后可能分配到不同的 partition 中
  所以贤姆，為 topic 創(chuàng)建 partitions 時(shí)要預(yù)留足夠的個(gè)數(shù)。這樣當(dāng)將來(lái)負(fù)載變大時(shí)稳衬，可以方便的通過(guò)添加 consumers 來(lái)分流霞捡。

上面我們提到的都是一個(gè) consumer group 對(duì) topic 的消費(fèi)。很多情況下薄疚，同一個(gè) topic 的消息會(huì)有多個(gè)不同的應(yīng)用(user cases)感興趣（每一個(gè) use case 都能拿到該 topic 的所有 messages）碧信，這時(shí)就需要為不同的 user case 創(chuàng)建不同的 consumer group，即有多個(gè) consumer groups 消費(fèi)同一個(gè) topic街夭。

consumer group5.jpg

4.2 Partition Rebalance

發(fā)生一下情況時(shí)砰碴，需要對(duì) consumer group 進(jìn)行 partition rebalance:

• 當(dāng) consumer group 的消費(fèi)能力不足時(shí)，增加 consumer
• 當(dāng) consumer crash /network failure 時(shí)板丽，移除 consumer
• 管理員增加新的 partition

通過(guò)對(duì) partition rebalance 的支持呈枉，Kafka 具備了 high availability & scalability趁尼。但是正常情況下，盡量避免 partition rebalance猖辫。因?yàn)椋?/p>

• consumer 不能消費(fèi)消息酥泞，consumer group 會(huì)有一個(gè)短暫的不可用期
• 當(dāng) partition 從原先的 consumer 移到新的 consumer 時(shí)，原先 consumer 丟失了它當(dāng)前處理的狀態(tài)

設(shè)計(jì)階段啃憎，就要把由于 rebalance 引發(fā)的潛在的消息重復(fù)處理的情況考慮進(jìn)去芝囤。

4.3 Group coordinator & Group leader

關(guān)于 consumers 的維護(hù)，有兩個(gè)重要的概念：group coordinator & group leader:

1. group coordinator: 是特殊的 broker辛萍。consumers poll 消息 & commit 消費(fèi)消息記錄時(shí)悯姊，會(huì)發(fā)送 heartbeats 到 group coordinator 來(lái)同時(shí)告知自己的健康狀況。
2. group leader: 第一個(gè)加入 consumer group 的 consumer 就是該 group 的 group leader贩毕。group coordinator 會(huì)把 consumers 列表發(fā)給 group leader 來(lái)維護(hù)悯许。group leader 負(fù)責(zé) assign & reassign partitions。

如果 consumer crashed／network failure耳幢，長(zhǎng)時(shí)間沒(méi)有發(fā)送 heartbeats 到 group coordinator 時(shí)岸晦，group coordinator 會(huì)認(rèn)為該 consumer 失聯(lián)欧啤，并通知 group leader rebalance partition睛藻，group leader 將 rebalance 的結(jié)果通知 group coordinator，由 coordinator 來(lái)通知 consumers 新的 partitions 關(guān)系邢隧。

4.4 Poll loop

consumer 中的核心功能幾乎都在 consumer.poll() 方法中店印。poll(timeout) 通過(guò) timeout 參數(shù)控制 poll 的阻塞等待數(shù)據(jù)時(shí)間。如果 timeout = 0倒慧，則立即返回按摘，無(wú)論是否有新消息。timeout 的值是需要根據(jù)自身業(yè)務(wù)設(shè)置的纫谅。但是它不僅僅是從 broker 中讀取消息：

1. 初始調(diào)用 poll() 時(shí)炫贤，會(huì)去找 groupCoordinator & 加入 consumer group & 接收 partition assignment
2. poll 內(nèi)部負(fù)責(zé)處理 partition rebalance
3. 發(fā)送 heartbeat: 當(dāng) consumer 停止 poll() 時(shí)，會(huì)停止發(fā) heartbeat付秕，被 group coordinator 認(rèn)為 fail兰珍，把分配給它的 partitions 分配給 consumer group 中別的 consumer。

所以:

• consumer 需要持續(xù) poll 數(shù)據(jù)
• consumer 處理數(shù)據(jù)的過(guò)程要越快越好询吴，避免由于長(zhǎng)時(shí)間不發(fā) heartbeat掠河，引起宕機(jī)誤判

4.5 Commits and Offsets

Kafka 不像大多數(shù) JMS queue 那樣，broker 不主動(dòng)跟蹤 consumer 的 ack猛计，而是通過(guò) consumer 發(fā)起 commit 來(lái)更新最新的 offset 到 _consumer_offsets topic 中

consumer.poll()時(shí), broker 會(huì)返回還沒(méi)有消費(fèi)的消息記錄唠摹，消息中帶有自身的 offset。
consumer 通過(guò) commit 動(dòng)作發(fā)送一個(gè)帶有 partition offset 的 message 到 kafka broker 的特殊 topic(__consumer_offsets topic) 來(lái)更新 offset奉瘤。

consumer 什么時(shí)候 commit 該消息呢勾拉？

Automatic Commit

配置 enable.auto.commit=true，consumer 會(huì)每隔一個(gè) interval (默認(rèn)每隔5s)自動(dòng)提交 consumer 通過(guò) poll() 收到的最大的 offset。
automatic commits 也是通過(guò) poll loop 來(lái)實(shí)現(xiàn)的藕赞。每次 poll, consumer 都會(huì)自動(dòng)檢查是否到時(shí)間執(zhí)行一次 commit 來(lái)提交最近一次 poll() 獲得的最大的 offsets苛秕。

當(dāng) consumer crashes / new consumer 加入 consumer group，會(huì)觸發(fā) rebalance找默。在 rebalance 后艇劫，每個(gè) consumer 被分配一組新的 partitions，并獲取到最新的 committed offset of each partition 來(lái)繼續(xù)工作惩激。
但是考慮以下情況：假設(shè)配置每隔 5s commit the latest offset店煞，上次提交 2s 后，發(fā)生了 rebalance风钻，所有 consumers 獲取到之前最近的 offsets顷蟀，但這個(gè) offset 其實(shí)是 2s 前的，這 2s 間到達(dá) consumers 的消息將會(huì)被處理兩次骡技。
可以配置較小的 interval 來(lái)減少重復(fù)消費(fèi)的消息鸣个，但是本質(zhì)上無(wú)法完全避免。

可以看出布朦，操作 committed offset 的位置囤萤，是可能發(fā)生以下情況：

• 重復(fù)讀取 & 消費(fèi)消息
• 遺漏處理消息

Commit Current Offset

如果想對(duì) offset 的控制更準(zhǔn)確，配置 auto.commit.offset=false是趴，手動(dòng) commit offset涛舍。

需要開(kāi)發(fā)人員手動(dòng)調(diào)用 commitSync()，將把 poll() 返回的最新的 offset唆途，建議:

• 在 client 處理完 poll 回的所有數(shù)據(jù)后富雅，再執(zhí)行該方法，否則會(huì)冒著丟失消息的風(fēng)險(xiǎn)肛搬。
• 面對(duì) rebalance 時(shí)没佑，仍然存在重復(fù)消費(fèi)消息的情況

Asynchronous Commit

commitSync() 會(huì)阻塞應(yīng)用，直到收到 broker 的明確響應(yīng)温赔。這將很大的影響系統(tǒng)吞吐蛤奢。可以考慮使用異步 consumer.commitAsync()

commitSync() 內(nèi)部有 retry让腹，如果遇到 retriable failure远剩，會(huì)持續(xù)重試，影響性能骇窍，如果遇到 nonretriable failure瓜晤，則會(huì)直接 commit fail。// todo retry count
commitAsync() 沒(méi)有 retry腹纳，之所以不支持 retry痢掠，是因?yàn)樗旧硎?async 的驱犹、非阻塞的。如果失敗了又重試足画，可能會(huì)把這段時(shí)間發(fā)生的更新的 commit 的數(shù)據(jù)修改回去雄驹。當(dāng)然了，如果真想重試淹辞，是可以找到解決方案的医舆，如記錄一個(gè)全局單調(diào)遞增的 sequence number，重試前檢查如果 offset 小于該 number象缀，則取消 retry蔬将。

Combining Synchronous and Asynchronous Commits

對(duì)于手動(dòng)控制 offset 的情況，commitAsync() & commitSync() 可以結(jié)合使用央星。正常情況下使用 async霞怀，提高性能，并且偶爾由于網(wǎng)絡(luò)原因發(fā)生的失敗也不需要 retry莉给，一般都會(huì)在接下來(lái)的 commit 中成功毙石，等待服務(wù)停止消費(fèi)時(shí)，調(diào)用 sync颓遏，確保最終正確提交 offset徐矩。

Commit Specified Offset

commitSync() & commitAsync() 存在一個(gè)問(wèn)題，只能在對(duì) batch messages 全部處理完后州泊，將最大的 offset 提交丧蘸，無(wú)法做到更細(xì)粒度的控制。當(dāng)處理 batch messages 的耗時(shí)很長(zhǎng)遥皂，或者 batch 的消息個(gè)數(shù)很多時(shí)，如果在消費(fèi)過(guò)程中發(fā)生了 rebalance刽漂，這次 poll 獲取的所有 messages 都需要重新處理一次演训。

commitSync() & commitAsync() 都提供了帶參數(shù)的方法，允許我們根據(jù)業(yè)務(wù)在消費(fèi) batch messages 的過(guò)程中按需要提交 offset贝咙。

4.6 Rebalance Listeners

在發(fā)生 partition rebalance 時(shí)(可能處在 processing batch messages 間隙)样悟，consumer 需要做一些 cleanup work，包括對(duì)正在處理的消息的收尾工作庭猩，對(duì)文件窟她、數(shù)據(jù)庫(kù)連接等的管理。我們可以通過(guò)在調(diào)用 consumer.subscribe() 方法中傳入自定義的 ConsumerRebalanceListener 來(lái)實(shí)現(xiàn)蔼水。

ConsumerRebalanceListener 有兩個(gè)方法需要實(shí)現(xiàn)：

• onPartitionsRevoked: 執(zhí)行 rebalance 前的收尾工作震糖，consumer 停止處理舊 messages 后 & rebalance 發(fā)生前系統(tǒng)調(diào)用該方法∨恳福可以通過(guò)該方法來(lái) commit client 已處理的 offsets吊说，這樣能 commit 準(zhǔn)確的 offsets
• onPartitionsAssigned: 執(zhí)行 rebalance 后的初始化工作论咏，partitions reassigned 后 & consumer 開(kāi)始處理新 messages 前調(diào)用該方法

Consuming Records with Specific Offsets

seek(TopicPartition partition, long offset)
seekToBeginning(TopicPartition tp)
seekToEnd(TopicPartition tp)

Standalone Consumer: Use a Consumer Without a Group

有些情況下，consumer 會(huì)需要指定消費(fèi)某些具體的 partitions, 而不是 join consumer group颁井，由 consumer group 分配 partition & rebalance厅贪。可以調(diào)用 consumer.assign() 來(lái)實(shí)現(xiàn)該需求雅宾。