摘要：學(xué)習(xí)筆記使用一個叫的文學(xué)家的名字用來命名的。引入，正式升級為分布式流處理平臺。主要還是針對組成員數(shù)量減少的情況。當(dāng)所有成員都退出組后，消費(fèi)者組狀態(tài)變更為。自動定期刪除過期位移的條件就是，組要處于狀態(tài)。減少下游系統(tǒng)一次性消費(fèi)的消息總數(shù)。

Kafka使用一個叫Franz Kafka的文學(xué)家的名字用來命名的。

Kafka是一款開源的消息引擎系統(tǒng)。也是一個分布式流處理平臺。

Kafka同時支持點對點模型以及發(fā)布/訂閱模型。

為什么要使用Kakfa？四個字：削峰填谷！

Record:消息，指Kafka處理對象

Topic:主題，用來承載消息的容器

Partition:分區(qū)，一個有序不變的消息隊列，一個主題下可以有多個分區(qū)

Offset:消息位移，表示分區(qū)中每條信息的位置，是一個單調(diào)遞增不變的值

Replica,副本，數(shù)據(jù)冗余。

領(lǐng)導(dǎo)者副本：對外提供服務(wù)，與客戶端進(jìn)行交互

追隨者副本：不能與外界進(jìn)行交互，只是被動地追隨領(lǐng)導(dǎo)者副本

Producer:生產(chǎn)者，向主題發(fā)布新消息的應(yīng)用程序

Consumer:消費(fèi)者，向主題訂閱新消息的應(yīng)用程序

Consumer Offset:消費(fèi)者位移，表示消費(fèi)者消費(fèi)進(jìn)度

Consumer Group:消費(fèi)者組，多個消費(fèi)者實例共同組成的一個組，同時消費(fèi)多個分區(qū)來實現(xiàn)高吞吐。

Rebalance:重平衡，消費(fèi)者組內(nèi)某個消費(fèi)者實例掛掉后，其他消費(fèi)者實例自動重新分配訂閱主題分區(qū)的過程。它是Kafka消費(fèi)者端實現(xiàn)高可用的重要手段。

Apache Kafka: 也稱社區(qū)版Kafka，迭代速度快，社區(qū)響應(yīng)度高，使用它可以讓你有更高的把控度；缺陷在于僅僅提供基礎(chǔ)核心組件，缺失一些高級特性

Confluent Kafka: 優(yōu)勢在于集成了很多高級特性且由Kafka原班人馬打造，質(zhì)量保證；缺陷在于國內(nèi)相關(guān)資料不全，普及率較低，沒有太多可參考的范例。

CDH/HDP Kafka: 優(yōu)勢在于操作簡單，節(jié)省運(yùn)維成本；缺陷在于把控度低，演進(jìn)速度慢

Kafka新版本客戶端代碼開始完全由java語言編寫，于是有些人開始“JAVA VS SCALA”的大討論。并從語言特性上分析為什么社區(qū)擯棄Scala轉(zhuǎn)而投向Java的懷抱。

其實事情沒有那么復(fù)雜，僅僅是因為社區(qū)來了一批Java程序猿，而以前老的scala程序猿隱退了罷了。

Kafka總共演進(jìn)了7個大版本

0.7版本： 上古版本，一旦有人向你推薦這個版本，懟他。

0.8版本： 開始引入副本機(jī)制，另外老版本需要制定zookeeper地址而不是Broker地址。在0.8.2.0版本社區(qū)引入了新版本Producer API,即指定Broker地址的Producer。

0.9版本： 重量級的大版本更迭。增加了基礎(chǔ)的安全認(rèn)證/權(quán)限功能，引入了Kafka Connect,新版本Producer API穩(wěn)定。

0.10.0.0: 里程碑的大版本。該版本又有兩個小版本，0.10.1和0.10.2。引入Kafka streams，正式升級為分布式流處理平臺。0.10.2.2 新Consumer API穩(wěn)定。

0.11.0.0: 目前最主流的版本之一。引入兩個重量級功能變更：一個是提供冪等性Producer API以及事務(wù) API, 另一個是對Kafka消息格式做了重構(gòu)。

1.0和2.0: 如果你是Kafka Stream用戶，至少選擇2.0.0版本吧。

最后還有個建議，不論你使用的是哪個版本，都請盡量保持服務(wù)端版本和客戶端版本一致，否則你將損失很多Kafka為你提供的性能優(yōu)化收益。

江湖經(jīng)驗：不要輕易成為新版本的小白鼠。

磁盤容量舉例：

假設(shè)公司有個業(yè)務(wù)需要每天向Kafka集群發(fā)送 1 億條信息。每條消息保存兩份來防止數(shù)據(jù)丟失。消息默認(rèn)保存兩周時間。并假設(shè)消息的平均大小是1KB。問你的Kafka集群需要為這個業(yè)務(wù)預(yù)留多少磁盤空間？

總大小：1億 1KB 2備份 * 14 ~= 2800G
加上Kafka的一些索引數(shù)據(jù)，為它預(yù)留10%，那么總大小變?yōu)?2800 * （1 + 10%） ~= 3TB

Kafka支持?jǐn)?shù)據(jù)壓縮，壓縮比0.75的話，那么應(yīng)該預(yù)留的存儲空間為2.25TB左右。

帶寬舉例

與其說是帶寬資源的規(guī)劃，其實真正要規(guī)劃的是Kafka服務(wù)器的數(shù)量。

假設(shè)公司機(jī)房環(huán)境1Gbps,現(xiàn)有個業(yè)務(wù)，需要在1小時內(nèi)處理1TB的業(yè)務(wù)數(shù)據(jù)。

一般單臺服務(wù)器 規(guī)劃使用70%的帶寬資源的1/3 ~= 240Mbps。

1TB需要1小時處理，則每秒差不多需要處理2336Mbps的數(shù)據(jù)，除 240Mbps，則差不多需要10臺機(jī)器。如果消息還需要額外復(fù)制的話，那么還要對應(yīng)乘上備份數(shù)。

壓縮配置

compression.type

壓縮算法

總結(jié)一下壓縮和解壓縮，Producer端壓縮，Broker端保持，Consumer端解壓縮。

不要使用producer.send(msg),而要使用producer.send(msg,callback)

設(shè)置acks=all,表明所有副本Broker都要接受消息，該消息才算是“已提交”

設(shè)置retries>0,表明Producer自動重試，當(dāng)網(wǎng)絡(luò)順斷時，防止消息丟失。

設(shè)置unclean.leader.election.enable=false

設(shè)置replication.factor >=3，增加副本數(shù)，保證數(shù)據(jù)冗余

設(shè)置min.insync.replicas > 1,控制的是消息至少要被寫入多少個副本才算是 已提交。

確保replication.factor > min.insync.replicas。如果兩者相等，那么只要有一個副本掛機(jī)，整個分區(qū)就無法正常工作了。推薦設(shè)置replication.factor = min.insync.replicas + 1

確保消息消費(fèi)完再提交。設(shè)置enable.aoto.commit=false

分為生產(chǎn)者攔截器和消費(fèi)者攔截器。

典型的應(yīng)用場景可以應(yīng)用于客戶端監(jiān)控、端到端系統(tǒng)性能測試、消息審計等多種功能在內(nèi)的場景。

KafkaProducer實例創(chuàng)建時啟動Sender線程，從而創(chuàng)建與bootstrap.servers中所有的Broker的TCP連接。

KafkaProducer實例首次更新元數(shù)據(jù)信息之后，還會再次創(chuàng)建與集群中所有Broker的TCP連接

如果Producer端發(fā)送信息到某臺Broker時，發(fā)現(xiàn)沒有與該Broker的TCP連接，那么也會創(chuàng)建連接

如果設(shè)置connections.max.idle.ms > 0,則步驟一中的TCP連接會被自動關(guān)閉；如果設(shè)置該參數(shù)-1，那么步驟一中創(chuàng)建的連接無法被關(guān)閉，會成為僵尸進(jìn)程。

創(chuàng)建的3個時機(jī)

發(fā)起FindCoordinator請求時

連接協(xié)調(diào)者時

消費(fèi)數(shù)據(jù)時

消費(fèi)者程序會創(chuàng)建3類TCP連接

確定協(xié)調(diào)者和獲取集群元數(shù)據(jù)

連接協(xié)調(diào)者，令其執(zhí)行組成員管理操作

執(zhí)行實際的消息獲取

消息交付可靠性保障，常見的承諾有以下三種

最多一次：消息可能會丟失，但絕不會重復(fù)發(fā)送

至少一次：消息不會丟失，但有可能被重復(fù)發(fā)送

精確一次：消息不會丟失，也不會被重復(fù)發(fā)送

Kafka默認(rèn)是最少一次

要保證精確一次，就需要冪等和事務(wù)。不過性能會想對較差。

冪等性有很多好處。其最大的優(yōu)勢在于我們可以安全地重試任何冪等性操作，反正它們不會破壞我們的系統(tǒng)狀態(tài)。

在0.11.0.0版本引入了冪等生產(chǎn)者，只要更改配置props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true)。

使用冪等生產(chǎn)者要注意

它只能保證單分區(qū)的冪等，多分區(qū)無法實現(xiàn)

只能實現(xiàn)單會話上的冪等，重啟之后冪等消失

設(shè)置事務(wù)型Producer

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true)

設(shè)置producer端參數(shù)transctional.id。最好為其設(shè)置一個有意義的名字

此外代碼也要做一些調(diào)整變化。

producer.initTransactions();
try {
            producer.beginTransaction();
            producer.send(record1);
            producer.send(record2);
            producer.commitTransaction();
} catch (KafkaException e) {
            producer.abortTransaction();
}

Rebalance發(fā)生的時機(jī)有三個

組成員數(shù)據(jù)量發(fā)生變化

訂閱主題數(shù)量發(fā)生變化

訂閱主題的分區(qū)數(shù)發(fā)生變化

后面兩個通常是運(yùn)維的主動操作，無法避免。主要還是針對組成員數(shù)量減少的情況。增加一般也是人為主動的。

那么避免因為參數(shù)或邏輯不合理而導(dǎo)致的成員退出，與之相關(guān)的主要參數(shù)

session.timeout.ms,推薦設(shè)置6s

heartbeat.interval.ms,推薦設(shè)置2s

max.poll.interval.ms,推薦設(shè)置比你的業(yè)務(wù)邏輯處理要長

GC參數(shù)，避免頻繁的FULL GC

重平衡過程是通過 消費(fèi)者端的心跳線程來通知到其他消費(fèi)者實例的。

0.10.1.0版本之前，發(fā)送心跳請求是在消費(fèi)者主線程完成的，也就是kafkaConsumer.poll方法的那個線程。這樣做有諸多弊端，因為消息處理也是在這個線程中完成的。因此當(dāng)業(yè)務(wù)邏輯處理消耗了較長時間，心跳請求就無法及時發(fā)送到協(xié)調(diào)者那邊了。導(dǎo)致協(xié)調(diào)者 錯誤地認(rèn)為該消費(fèi)者已經(jīng)死了。

0.10.1.0版本開始，社區(qū)引入了一個多帶帶的線程來專門執(zhí)行心跳發(fā)送。

定義了5種狀態(tài)

各個狀態(tài)的流轉(zhuǎn)

一個消費(fèi)者組最開始是Empty狀態(tài)，當(dāng)重平衡過程開啟后，它會被置為PreparingRebalance狀態(tài)等待成員加入，之后變更到CompletingRebalance狀態(tài)等待分配方案，最后流轉(zhuǎn)到Stable狀態(tài)完成重平衡。

當(dāng)有新成員或已有成員退出時，消費(fèi)者組的狀態(tài)從Stable直接跳到PreparingRebalance狀態(tài)，此時，所有現(xiàn)存成員就必須重新申請加入組。

當(dāng)所有成員都退出組后，消費(fèi)者組狀態(tài)變更為Empty。

Kafka自動定期刪除過期位移的條件就是，組要處于Empty狀態(tài)。

JoinGroup請求

SyncGroup請求

新成員入組

組成員主動離組

組成員崩潰離組

重平衡時協(xié)調(diào)者對組內(nèi)成員提交位移的處理

縮短消息處理的時間，該方法優(yōu)先處理

增加Consumer端允許下游系統(tǒng)消費(fèi)一批數(shù)據(jù)的最大時長。設(shè)置參數(shù)max.poll.interval.ms，新版本默認(rèn)是5分鐘。

減少下游系統(tǒng)一次性消費(fèi)的消息總數(shù)。max.poll.records

下游系統(tǒng)使用多線程來加速消費(fèi)

鑒于KafkaConsumer不是線程安全的事實，制定兩套多線程方案。

每個線程維護(hù)專屬的KafkaConsumer實例，負(fù)責(zé)完整的消息獲取、消息處理流程

核心代碼

```
public class KafkaConsumerRunner implements Runnable {
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final KafkaConsumer consumer;


     public void run() {
         try {
             consumer.subscribe(Arrays.asList("topic"));
             while (!closed.get()) {
            ConsumerRecords records = 
                consumer.poll(Duration.ofMillis(10000));
                 //  執(zhí)行消息處理邏輯
             }
         } catch (WakeupException e) {
             // Ignore exception if closing
             if (!closed.get()) throw e;
         } finally {
             consumer.close();
         }
     }


     // Shutdown hook which can be called from a separate thread
     public void shutdown() {
         closed.set(true);
         consumer.wakeup();
     }
```

消費(fèi)者程序使用單或多線程獲取消息，創(chuàng)建多個消費(fèi)者線程執(zhí)行消息處理邏輯

核心代碼

```
private final KafkaConsumer consumer;
private ExecutorService executors;
...


private int workerNum = ...;
executors = new ThreadPoolExecutor(
    workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<>(1000), 
    new ThreadPoolExecutor.CallerRunsPolicy());


...
while (true)  {
    ConsumerRecords records = 
        consumer.poll(Duration.ofSeconds(1));
    for (final ConsumerRecord record : records) {
        executors.submit(new Worker(record));
    }
}

```

兩種方案各有特點。

使用Kafka自帶命令行工具kafka-consumer-groups腳本

使用Kafka Consumer API

使用Kafka自帶的JMX監(jiān)控指標(biāo)

副本機(jī)制的好處：

提供數(shù)據(jù)冗余

提供高伸縮性

改善數(shù)據(jù)局部性

但Kafka只有第一種好處，原因是這樣的設(shè)計，Kafka有兩點好處

方便實現(xiàn) Read-your-writes

指當(dāng)你用生產(chǎn)者API向Kafka成功寫入消息后，馬上使用消費(fèi)者API去讀取剛才生產(chǎn)的消息

方便實現(xiàn)單調(diào)讀（Monotonic Reads）

在多次消費(fèi)信息時，不會看到該消息一會存在一會不存在的情況。

判斷Follower副本與Leader副本是否同步的標(biāo)準(zhǔn)，Broker參數(shù)replia.lag.time.max.ms的參數(shù)值。Kafka有一個in-sync Replicas(ISR）集合的概念。

控制器組件（Controller）,是Kafka的核心組件，它的主要作用是在Apache Zookeeper的幫助下管理和協(xié)調(diào)整個Kafka集群。

每臺Broker都能充當(dāng)控制器，在Broker啟動時，會嘗試去Zookeeper中創(chuàng)建/controller節(jié)點。Kafka當(dāng)前選舉規(guī)則，第一個成功創(chuàng)建/controller節(jié)點的Broker會被指定為控制器。

主題管理

分區(qū)重分配

Prefered領(lǐng)導(dǎo)者選舉

集群成員管理

數(shù)據(jù)服務(wù)，控制器上保存最全的集群元數(shù)據(jù)信息

這些數(shù)據(jù)其實也在Zookeeper中存儲了一份。

小竅門分享：當(dāng)你覺得控制器出現(xiàn)問題時，比如主題無法刪除了，重分區(qū)hang住了，你可以不用重啟broker或者控制器，快速簡便的方法，直接去Zookeeper手動刪除/controller節(jié)點。

這樣做的好處是，既可以引發(fā)控制器的重選舉，又可以避免重啟Broker導(dǎo)致的消息中斷。

Kafka方案類似于Reactor模式

那么Kafka類似的方案是這樣的。網(wǎng)絡(luò)線程池默認(rèn)參數(shù)num.network.threads=3

好了，客戶端發(fā)來的請求會被Aceptor線程分發(fā)到任意一個網(wǎng)絡(luò)線程中，由他們進(jìn)行處理。你可能會認(rèn)為，網(wǎng)絡(luò)線程池是順序處理不就好了？實際上，Kafka在這個環(huán)節(jié)上又做了一層異步線程池的處理。

IO線程池執(zhí)行真正的處理。如果是PRODUCER生產(chǎn)請求，則將消息寫入到底層的磁盤日志中；如果是FETCH請求，則從磁盤或頁緩存中讀取消息。當(dāng)IO請求處理完請求后，會將生成的響應(yīng)放入網(wǎng)絡(luò)線程池的響應(yīng)隊列中，并由對應(yīng)的網(wǎng)絡(luò)線程負(fù)責(zé)將Response反還給客戶端。

請求隊列是所有網(wǎng)絡(luò)線程共享的，而響應(yīng)隊列則是每個網(wǎng)絡(luò)線程專屬的。

IO線程池默認(rèn)參數(shù)num.io.threads=8

圖中還有一個Purgatory的組件，這是Kafka中著名的“煉獄”組件。

它是用來緩存延時請求的，所謂延時請求，就是那些一時未滿足條件的不可立刻處理的請求。