什么是 Kafka

Kafka 是由 Linkedin 公司開發的，它是一個分布式的，支持多分區、多副本，基于 Zookeeper 的分布式消息流平臺，它同時也是一款開源的基于發布訂閱模式的消息引擎系統。

Kafka 的基本術語

消息：Kafka 中的數據單元被稱為消息，也被稱為記錄，可以把它看作數據庫表中某一行的記錄。

批次：為了提高效率， 消息會分批次寫入 Kafka，批次就代指的是一組消息。

主題：消息的種類稱為 主題（Topic）,可以說一個主題代表了一類消息。相當于是對消息進行分類。主題就像是數據庫中的表。

分區：主題可以被分為若干個分區（partition），同一個主題中的分區可以不在一個機器上，有可能會部署在多個機器上，由此來實現 kafka 的伸縮性，單一主題中的分區有序，但是無法保證主題中所有的分區有序

生產者： 向主題發布消息的客戶端應用程序稱為生產者（Producer），生產者用于持續不斷的向某個主題發送消息。

消費者：訂閱主題消息的客戶端程序稱為消費者（Consumer），消費者用于處理生產者產生的消息。

消費者群組：生產者與消費者的關系就如同餐廳中的廚師和顧客之間的關系一樣，一個廚師對應多個顧客，也就是一個生產者對應多個消費者，消費者群組（Consumer Group）指的就是由一個或多個消費者組成的群體。

偏移量：偏移量（Consumer Offset）是一種元數據，它是一個不斷遞增的整數值，用來記錄消費者發生重平衡時的位置，以便用來恢復數據。

broker: 一個獨立的 Kafka 服務器就被稱為 broker，broker 接收來自生產者的消息，為消息設置偏移量，并提交消息到磁盤保存。

broker 集群：broker 是集群 的組成部分，broker 集群由一個或多個 broker 組成，每個集群都有一個 broker 同時充當了集群控制器的角色（自動從集群的活躍成員中選舉出來）。

副本：Kafka 中消息的備份又叫做 副本（Replica），副本的數量是可以配置的，Kafka 定義了兩類副本：領導者副本（Leader Replica） 和 追隨者副本（Follower Replica），前者對外提供服務，后者只是被動跟隨。

重平衡：Rebalance。消費者組內某個消費者實例掛掉后，其他消費者實例自動重新分配訂閱主題分區的過程。Rebalance 是 Kafka 消費者端實現高可用的重要手段。

Kafka 的特性（設計原則）

• 高吞吐、低延遲：kakfa 最大的特點就是收發消息非常快，kafka 每秒可以處理幾十萬條消息，它的最低延遲只有幾毫秒。
• 高伸縮性： 每個主題(topic) 包含多個分區(partition)，主題中的分區可以分布在不同的主機(broker)中。
• 持久性、可靠性： Kafka 能夠允許數據的持久化存儲，消息被持久化到磁盤，并支持數據備份防止數據丟失，Kafka 底層的數據存儲是基于 Zookeeper 存儲的，Zookeeper 我們知道它的數據能夠持久存儲。
• 容錯性： 允許集群中的節點失敗，某個節點宕機，Kafka 集群能夠正常工作
• 高并發： 支持數千個客戶端同時讀寫

Kafka 的使用場景

• 活動跟蹤：Kafka 可以用來跟蹤用戶行為，比如我們經常回去淘寶購物，你打開淘寶的那一刻，你的登陸信息，登陸次數都會作為消息傳輸到 Kafka ，當你瀏覽購物的時候，你的瀏覽信息，你的搜索指數，你的購物愛好都會作為一個個消息傳遞給 Kafka ，這樣就可以生成報告，可以做智能推薦，購買喜好等。
• 傳遞消息：Kafka 另外一個基本用途是傳遞消息，應用程序向用戶發送通知就是通過傳遞消息來實現的，這些應用組件可以生成消息，而不需要關心消息的格式，也不需要關心消息是如何發送的。
• 度量指標：Kafka也經常用來記錄運營監控數據。包括收集各種分布式應用的數據，生產各種操作的集中反饋，比如報警和報告。
• 日志記錄：Kafka 的基本概念來源于提交日志，比如我們可以把數據庫的更新發送到 Kafka 上，用來記錄數據庫的更新時間，通過kafka以統一接口服務的方式開放給各種consumer，例如hadoop、Hbase、Solr等。
• 流式處理：流式處理是有一個能夠提供多種應用程序的領域。
• 限流削峰：Kafka 多用于互聯網領域某一時刻請求特別多的情況下，可以把請求寫入Kafka 中，避免直接請求后端程序導致服務崩潰。

Kafka 的消息隊列

Kafka 的消息隊列一般分為兩種模式：點對點模式和發布訂閱模式

Kafka 是支持消費者群組的，也就是說 Kafka 中會有一個或者多個消費者，如果一個生產者生產的消息由一個消費者進行消費的話，那么這種模式就是點對點模式

如果一個生產者或者多個生產者產生的消息能夠被多個消費者同時消費的情況，這樣的消息隊列成為發布訂閱模式的消息隊列

Kafka 系統架構

如上圖所示，一個典型的 Kafka 集群中包含若干Producer（可以是web前端產生的Page View，或者是服務器日志，系統CPU、Memory等），若干broker（Kafka支持水平擴展，一般broker數量越多，集群吞吐率越高），若干Consumer Group，以及一個Zookeeper集群。Kafka通過Zookeeper管理集群配置，選舉leader，以及在Consumer Group發生變化時進行rebalance。Producer使用push模式將消息發布到broker，Consumer使用pull模式從broker訂閱并消費消息。

核心 API

Kafka 有四個核心API，它們分別是

• Producer API，它允許應用程序向一個或多個 topics 上發送消息記錄
• Consumer API，允許應用程序訂閱一個或多個 topics 并處理為其生成的記錄流
• Streams API，它允許應用程序作為流處理器，從一個或多個主題中消費輸入流并為其生成輸出流，有效的將輸入流轉換為輸出流。
• Connector API，它允許構建和運行將 Kafka 主題連接到現有應用程序或數據系統的可用生產者和消費者。例如，關系數據庫的連接器可能會捕獲對表的所有更改

Kafka 為何如此之快

Kafka 實現了零拷貝原理來快速移動數據，避免了內核之間的切換。Kafka 可以將數據記錄分批發送，從生產者到文件系統（Kafka 主題日志）到消費者，可以端到端的查看這些批次的數據。

批處理能夠進行更有效的數據壓縮并減少 I/O 延遲，Kafka 采取順序寫入磁盤的方式，避免了隨機磁盤尋址的浪費，更多關于磁盤尋址的了解，請參閱 程序員需要了解的硬核知識之磁盤 。

總結一下其實就是四個要點

• 順序讀寫
• 零拷貝
• 消息壓縮
• 分批發送

Kafka 重要配置。

來說一下 Kafka 中的重要參數配置吧，這些參數對 Kafka 來說是非常重要的。

broker 端配置

• broker.id
  每個 kafka broker 都有一個唯一的標識來表示，這個唯一的標識符即是 broker.id，它的默認值是 0。這個值在 kafka 集群中必須是唯一的，這個值可以任意設定，

• port
  如果使用配置樣本來啟動 kafka，它會監聽 9092 端口。修改 port 配置參數可以把它設置成任意的端口。要注意，如果使用 1024 以下的端口，需要使用 root 權限啟動 kakfa。

• zookeeper.connect
  用于保存 broker 元數據的 Zookeeper 地址是通過 zookeeper.connect 來指定的。比如我可以這么指定 localhost:2181 表示這個 Zookeeper 是運行在本地 2181 端口上的。我們也可以通過 比如我們可以通過 zk1:2181,zk2:2181,zk3:2181 來指定 zookeeper.connect 的多個參數值。該配置參數是用冒號分割的一組 hostname:port/path 列表，其含義如下

hostname 是 Zookeeper 服務器的機器名或者 ip 地址。port 是 Zookeeper 客戶端的端口號/path 是可選擇的 Zookeeper 路徑，Kafka 路徑是使用了 chroot 環境，如果不指定默認使用跟路徑。如果你有兩套 Kafka 集群，假設分別叫它們 kafka1 和 kafka2，那么兩套集群的zookeeper.connect參數可以這樣指定：zk1:2181,zk2:2181,zk3:2181/kafka1和zk1:2181,zk2:2181,zk3:2181/kafka2

• log.dirs
  Kafka 把所有的消息都保存到磁盤上，存放這些日志片段的目錄是通過 log.dirs 來制定的，它是用一組逗號來分割的本地系統路徑，log.dirs 是沒有默認值的，你必須手動指定他的默認值。其實還有一個參數是 log.dir，如你所知，這個配置是沒有 s 的，默認情況下只用配置 log.dirs 就好了，比如你可以通過 /home/kafka1,/home/kafka2,/home/kafka3 這樣來配置這個參數的值。

• num.recovery.threads.per.data.dir
  對于如下3種情況，Kafka 會使用可配置的線程池來處理日志片段。

服務器正常啟動，用于打開每個分區的日志片段；

服務器崩潰后重啟，用于檢查和截斷每個分區的日志片段；

服務器正常關閉，用于關閉日志片段。

默認情況下，每個日志目錄只使用一個線程。因為這些線程只是在服務器啟動和關閉時會用到，所以完全可以設置大量的線程來達到井行操作的目的。特別是對于包含大量分區的服務器來說，一旦發生崩憤，在進行恢復時使用井行操作可能會省下數小時的時間。設置此參數時需要注意，所配置的數字對應的是 log.dirs 指定的單個日志目錄。也就是說，如果 num.recovery.threads.per.data.dir 被設為 8，并且 log.dir 指定了 3 個路徑，那么總共需要 24 個線程。

• auto.create.topics.enable
  默認情況下，kafka 會使用三種方式來自動創建主題，下面是三種情況：

當一個生產者開始往主題寫入消息時

當一個消費者開始從主題讀取消息時

當任意一個客戶端向主題發送元數據請求時

auto.create.topics.enable參數我建議最好設置成 false，即不允許自動創建 Topic。在我們的線上環境里面有很多名字稀奇古怪的 Topic，我想大概都是因為該參數被設置成了 true 的緣故。

主題默認配置

Kafka 為新創建的主題提供了很多默認配置參數，下面就來一起認識一下這些參數

• num.partitions
  num.partitions 參數指定了新創建的主題需要包含多少個分區。如果啟用了主題自動創建功能（該功能是默認啟用的），主題分區的個數就是該參數指定的值。該參數的默認值是 1。要注意，我們可以增加主題分區的個數，但不能減少分區的個數。

• default.replication.factor
  這個參數比較簡單，它表示 kafka保存消息的副本數，如果一個副本失效了，另一個還可以繼續提供服務default.replication.factor 的默認值為1，這個參數在你啟用了主題自動創建功能后有效。

• log.retention.ms
  Kafka 通常根據時間來決定數據可以保留多久。默認使用 log.retention.hours 參數來配置時間，默認是 168 個小時，也就是一周。除此之外，還有兩個參數 log.retention.minutes 和 log.retentiion.ms 。這三個參數作用是一樣的，都是決定消息多久以后被刪除，推薦使用 log.retention.ms。

• log.retention.bytes
  另一種保留消息的方式是判斷消息是否過期。它的值通過參數 log.retention.bytes 來指定，作用在每一個分區上。也就是說，如果有一個包含 8 個分區的主題，并且 log.retention.bytes 被設置為 1GB，那么這個主題最多可以保留 8GB 數據。所以，當主題的分區個數增加時，整個主題可以保留的數據也隨之增加。

• log.segment.bytes
  上述的日志都是作用在日志片段上，而不是作用在單個消息上。當消息到達 broker 時，它們被追加到分區的當前日志片段上，當日志片段大小到達 log.segment.bytes 指定上限（默認為 1GB）時，當前日志片段就會被關閉，一個新的日志片段被打開。如果一個日志片段被關閉，就開始等待過期。這個參數的值越小，就越會頻繁的關閉和分配新文件，從而降低磁盤寫入的整體效率。

• log.segment.ms
  上面提到日志片段經關閉后需等待過期，那么 log.segment.ms 這個參數就是指定日志多長時間被關閉的參數和，log.segment.ms 和 log.retention.bytes 也不存在互斥問題。日志片段會在大小或時間到達上限時被關閉，就看哪個條件先得到滿足。

• message.max.bytes
  broker 通過設置 message.max.bytes 參數來限制單個消息的大小，默認是 1000 000， 也就是 1MB，如果生產者嘗試發送的消息超過這個大小，不僅消息不會被接收，還會收到 broker 返回的錯誤消息。跟其他與字節相關的配置參數一樣，該參數指的是壓縮后的消息大小，也就是說，只要壓縮后的消息小于 mesage.max.bytes，那么消息的實際大小可以大于這個值

這個值對性能有顯著的影響。值越大，那么負責處理網絡連接和請求的線程就需要花越多的時間來處理這些請求。它還會增加磁盤寫入塊的大小，從而影響 IO 吞吐量。

• retention.ms
  規定了該主題消息被保存的時常，默認是7天，即該主題只能保存7天的消息，一旦設置了這個值，它會覆蓋掉 Broker 端的全局參數值。

• retention.bytes
  retention.bytes：規定了要為該 Topic 預留多大的磁盤空間。和全局參數作用相似，這個值通常在多租戶的 Kafka 集群中會有用武之地。當前默認值是 -1，表示可以無限使用磁盤空間。

JVM 參數配置

JDK 版本一般推薦直接使用 JDK1.8，這個版本也是現在中國大部分程序員的首選版本。

說到 JVM 端設置，就繞不開堆這個話題，業界最推崇的一種設置方式就是直接將 JVM 堆大小設置為 6GB，這樣會避免很多 Bug 出現。

JVM 端配置的另一個重要參數就是垃圾回收器的設置，也就是平時常說的 GC 設置。如果你依然在使用 Java 7，那么可以根據以下法則選擇合適的垃圾回收器：

• 如果 Broker 所在機器的 CPU 資源非常充裕，建議使用 CMS 收集器。
  啟用方法是指定-XX:+UseCurrentMarkSweepGC。

• 否則，使用吞吐量收集器。開啟方法是指定-XX:+UseParallelGC。
  當然了，如果你已經在使用 Java 8 了，那么就用默認的 G1 收集器就好了。在沒有任何調優的情況下，G1 表現得要比 CMS 出色，主要體現在更少的 Full GC，需要調整的參數更少等，所以使用 G1 就好了。一般 G1 的調整只需要這兩個參數即可

• MaxGCPauseMillis
  該參數指定每次垃圾回收默認的停頓時間。該值不是固定的，G1可以根據需要使用更長的時間。它的默認值是 200ms，也就是說，每一輪垃圾回收大概需要200 ms 的時間。

• InitiatingHeapOccupancyPercent
  該參數指定了 G1 啟動新一輪垃圾回收之前可以使用的堆內存百分比，默認值是45，這就表明G1在堆使用率到達45之前不會啟用垃圾回收。這個百分比包括新生代和老年代。

Kafka Producer

在 Kafka 中，我們把產生消息的那一方稱為生產者，比如我們經常回去淘寶購物，你打開淘寶的那一刻，你的登陸信息，登陸次數都會作為消息傳輸到 Kafka 后臺，當你瀏覽購物的時候，你的瀏覽信息，你的搜索指數，你的購物愛好都會作為一個個消息傳遞給 Kafka 后臺，然后淘寶會根據你的愛好做智能推薦，致使你的錢包從來都禁不住誘惑，那么這些生產者產生的消息是怎么傳到 Kafka 應用程序的呢？發送過程是怎么樣的呢？

盡管消息的產生非常簡單，但是消息的發送過程還是比較復雜的，如圖

我們從創建一個ProducerRecord 對象開始，ProducerRecord 是 Kafka 中的一個核心類，它代表了一組 Kafka 需要發送的 key/value 鍵值對，它由記錄要發送到的主題名稱（Topic Name），可選的分區號（Partition Number）以及可選的鍵值對構成。

在發送 ProducerRecord 時，我們需要將鍵值對對象由序列化器轉換為字節數組，這樣它們才能夠在網絡上傳輸。然后消息到達了分區器。

如果發送過程中指定了有效的分區號，那么在發送記錄時將使用該分區。如果發送過程中未指定分區，則將使用key 的 hash 函數映射指定一個分區。如果發送的過程中既沒有分區號也沒有，則將以循環的方式分配一個分區。選好分區后，生產者就知道向哪個主題和分區發送數據了。

ProducerRecord 還有關聯的時間戳，如果用戶沒有提供時間戳，那么生產者將會在記錄中使用當前的時間作為時間戳。Kafka 最終使用的時間戳取決于 topic 主題配置的時間戳類型。

• 如果將主題配置為使用 CreateTime，則生產者記錄中的時間戳將由 broker 使用。
• 如果將主題配置為使用LogAppendTime，則生產者記錄中的時間戳在將消息添加到其日志中時，將由 broker 重寫。
  然后，這條消息被存放在一個記錄批次里，這個批次里的所有消息會被發送到相同的主題和分區上。由一個獨立的線程負責把它們發到 Kafka Broker 上。

Kafka Broker 在收到消息時會返回一個響應，如果寫入成功，會返回一個 RecordMetaData 對象，它包含了主題和分區信息，以及記錄在分區里的偏移量，上面兩種的時間戳類型也會返回給用戶。如果寫入失敗，會返回一個錯誤。生產者在收到錯誤之后會嘗試重新發送消息，幾次之后如果還是失敗的話，就返回錯誤消息。

創建 Kafka 生產者

要向 Kafka 寫入消息，首先需要創建一個生產者對象，并設置一些屬性。Kafka 生產者有3個必選的屬性

• bootstrap.servers
  該屬性指定 broker 的地址清單，地址的格式為 host:port。清單里不需要包含所有的 broker 地址，生產者會從給定的 broker 里查找到其他的 broker 信息。不過建議至少要提供兩個 broker 信息，一旦其中一個宕機，生產者仍然能夠連接到集群上。

• key.serializer
  broker 需要接收到序列化之后的 key/value 值，所以生產者發送的消息需要經過序列化之后才傳遞給 Kafka Broker。生產者需要知道采用何種方式把 Java 對象轉換為字節數組。key.serializer 必須被設置為一個實現了org.apache.kafka.common.serialization.Serializer 接口的類，生產者會使用這個類把鍵對象序列化為字節數組。這里拓展一下 Serializer 類

Serializer 是一個接口，它表示類將會采用何種方式序列化，它的作用是把對象轉換為字節，實現了 Serializer 接口的類主要有 ByteArraySerializer、StringSerializer、IntegerSerializer ，其中 ByteArraySerialize 是 Kafka 默認使用的序列化器，其他的序列化器還有很多，你可以通過 這里 查看其他序列化器。要注意的一點：key.serializer 是必須要設置的，即使你打算只發送值的內容。

• value.serializer
  與 key.serializer 一樣，value.serializer 指定的類會將值序列化。

下面代碼演示了如何創建一個 Kafka 生產者，這里只指定了必要的屬性，其他使用默認的配置

private Properties properties = new Properties(); properties.put("bootstrap.servers","broker1:9092,broker2:9092"); properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer"); properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); properties = new KafkaProducer<String,String>(properties);

來解釋一下這段代碼

• 首先創建了一個 Properties 對象
• 使用 StringSerializer 序列化器序列化 key / value 鍵值對
• 在這里我們創建了一個新的生產者對象，并為鍵值設置了恰當的類型，然后把 Properties 對象傳遞給他。

Kafka 消息發送

實例化生產者對象后，接下來就可以開始發送消息了，發送消息主要由下面幾種方式

簡單消息發送

Kafka 最簡單的消息發送如下：

ProducerRecord<String,String> record =new ProducerRecord<String, String>("CustomerCountry","West","France");producer.send(record);

代碼中生產者(producer)的 send() 方法需要把 ProducerRecord 的對象作為參數進行發送，ProducerRecord 有很多構造函數，這個我們下面討論，這里調用的是

public ProducerRecord(String topic, K key, V value) {}

這個構造函數，需要傳遞的是 topic主題，key 和 value。

把對應的參數傳遞完成后，生產者調用 send() 方法發送消息（ProducerRecord對象）。我們可以從生產者的架構圖中看出，消息是先被寫入分區中的緩沖區中，然后分批次發送給 Kafka Broker。

發送成功后，send() 方法會返回一個 Future(java.util.concurrent) 對象，Future 對象的類型是 RecordMetadata 類型，我們上面這段代碼沒有考慮返回值，所以沒有生成對應的 Future 對象，所以沒有辦法知道消息是否發送成功。如果不是很重要的信息或者對結果不會產生影響的信息，可以使用這種方式進行發送。

我們可以忽略發送消息時可能發生的錯誤或者在服務器端可能發生的錯誤，但在消息發送之前，生產者還可能發生其他的異常。這些異常有可能是 SerializationException(序列化失敗)，BufferedExhaustedException 或 TimeoutException(說明緩沖區已滿)，又或是 InterruptedException(說明發送線程被中斷)

同步發送消息

第二種消息發送機制如下所示

ProducerRecord<String,String> record =new ProducerRecord<String, String>("CustomerCountry","West","France");try{RecordMetadata recordMetadata = producer.send(record).get(); }catch(Exception e){e.printStackTrace()； }

這種發送消息的方式較上面的發送方式有了改進，首先調用 send() 方法，然后再調用 get() 方法等待 Kafka 響應。如果服務器返回錯誤，get() 方法會拋出異常，如果沒有發生錯誤，我們會得到 RecordMetadata 對象，可以用它來查看消息記錄。

生產者（KafkaProducer）在發送的過程中會出現兩類錯誤：其中一類是重試錯誤，這類錯誤可以通過重發消息來解決。比如連接的錯誤，可以通過再次建立連接來解決；無主錯誤則可以通過重新為分區選舉首領來解決。KafkaProducer 被配置為自動重試，如果多次重試后仍無法解決問題，則會拋出重試異常。另一類錯誤是無法通過重試來解決的，比如消息過大對于這類錯誤，KafkaProducer 不會進行重試，直接拋出異常。

異步發送消息

同步發送消息都有個問題，那就是同一時間只能有一個消息在發送，這會造成許多消息無法直接發送，造成消息滯后，無法發揮效益最大化。

比如消息在應用程序和 Kafka 集群之間一個來回需要 10ms。如果發送完每個消息后都等待響應的話，那么發送100個消息需要 1 秒，但是如果是異步方式的話，發送 100 條消息所需要的時間就會少很多很多。大多數時候，雖然Kafka 會返回 RecordMetadata 消息，但是我們并不需要等待響應。

為了在異步發送消息的同時能夠對異常情況進行處理，生產者提供了回掉支持。下面是回調的一個例子

ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("CustomerCountry", "Huston", "America");producer.send(producerRecord,new DemoProducerCallBack());class DemoProducerCallBack implements Callback {public void onCompletion(RecordMetadata metadata, Exception exception) {if(exception != null){exception.printStackTrace();;}} }

首先實現回調需要定義一個實現了org.apache.kafka.clients.producer.Callback的類，這個接口只有一個 onCompletion方法。如果 kafka 返回一個錯誤，onCompletion 方法會拋出一個非空(non null)異常，這里我們只是簡單的把它打印出來，如果是生產環境需要更詳細的處理，然后在 send() 方法發送的時候傳遞一個 Callback 回調的對象。

生產者分區機制

Kafka 對于數據的讀寫是以分區為粒度的，分區可以分布在多個主機（Broker）中，這樣每個節點能夠實現獨立的數據寫入和讀取，并且能夠通過增加新的節點來增加 Kafka 集群的吞吐量，通過分區部署在多個 Broker 來實現負載均衡的效果。

上面我們介紹了生產者的發送方式有三種：不管結果如何直接發送、發送并返回結果、發送并回調。由于消息是存在主題（topic）的分區（partition）中的，所以當 Producer 生產者發送產生一條消息發給 topic 的時候，你如何判斷這條消息會存在哪個分區中呢？

這其實就設計到 Kafka 的分區機制了。

分區策略

Kafka 的分區策略指的就是將生產者發送到哪個分區的算法。Kafka 為我們提供了默認的分區策略，同時它也支持你自定義分區策略。

如果要自定義分區策略的話，你需要顯示配置生產者端的參數 Partitioner.class，我們可以看一下這個類它位于 org.apache.kafka.clients.producer 包下

public interface Partitioner extends Configurable, Closeable {public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);public void close();default public void onNewBatch(String topic, Cluster cluster, int prevPartition) {} }

Partitioner 類有三個方法，分別來解釋一下

• partition(): 這個類有幾個參數: topic，表示需要傳遞的主題；key 表示消息中的鍵值；keyBytes表示分區中序列化過后的key，byte數組的形式傳遞；value 表示消息的 value 值；valueBytes 表示分區中序列化后的值數組；cluster表示當前集群的原數據。Kafka 給你這么多信息，就是希望讓你能夠充分地利用這些信息對消息進行分區，計算出它要被發送到哪個分區中。
• close() : 繼承了 Closeable 接口能夠實現 close() 方法，在分區關閉時調用。
• onNewBatch(): 表示通知分區程序用來創建新的批次
  其中與分區策略息息相關的就是 partition() 方法了，分區策略有下面這幾種

順序輪詢

順序分配，消息是均勻的分配給每個 partition，即每個分區存儲一次消息。就像下面這樣

上圖表示的就是輪詢策略，輪訓策略是 Kafka Producer 提供的默認策略，如果你不使用指定的輪訓策略的話，Kafka 默認會使用順序輪訓策略的方式。

隨機輪詢

隨機輪詢簡而言之就是隨機的向 partition 中保存消息，如下圖所示

實現隨機分配的代碼只需要兩行，如下

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); return ThreadLocalRandom.current().nextInt(partitions.size());

先計算出該主題總的分區數，然后隨機地返回一個小于它的正整數。

本質上看隨機策略也是力求將數據均勻地打散到各個分區，但從實際表現來看，它要遜于輪詢策略，所以如果追求數據的均勻分布，還是使用輪詢策略比較好。事實上，隨機策略是老版本生產者使用的分區策略，在新版本中已經改為輪詢了。

按照 key 進行消息保存

這個策略也叫做 key-ordering 策略，Kafka 中每條消息都會有自己的key，一旦消息被定義了 Key，那么你就可以保證同一個 Key 的所有消息都進入到相同的分區里面，由于每個分區下的消息處理都是有順序的，故這個策略被稱為按消息鍵保序策略，如下圖所示

實現這個策略的 partition 方法同樣簡單，只需要下面兩行代碼即可：

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); return Math.abs(key.hashCode()) % partitions.size();

上面這幾種分區策略都是比較基礎的策略，除此之外，你還可以自定義分區策略。

生產者壓縮機制

壓縮一詞簡單來講就是一種互換思想，它是一種經典的用 CPU 時間去換磁盤空間或者 I/O 傳輸量的思想，希望以較小的 CPU 開銷帶來更少的磁盤占用或更少的網絡 I/O 傳輸。。

Kafka 壓縮是什么

Kafka 的消息分為兩層：消息集合 和 消息。一個消息集合中包含若干條日志項，而日志項才是真正封裝消息的地方。Kafka 底層的消息日志由一系列消息集合日志項組成。Kafka 通常不會直接操作具體的一條條消息，它總是在消息集合這個層面上進行寫入操作。

在 Kafka 中，壓縮會發生在兩個地方：Kafka Producer 和 Kafka Consumer，為什么啟用壓縮？說白了就是消息太大，需要變小一點 來使消息發的更快一些。

Kafka Producer 中使用 compression.type 來開啟壓縮

private Properties properties = new Properties(); properties.put("bootstrap.servers","192.168.1.9:9092"); properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer"); properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); properties.put("compression.type", "gzip");Producer<String,String> producer = new KafkaProducer<String, String>(properties);ProducerRecord<String,String> record =new ProducerRecord<String, String>("CustomerCountry","Precision Products","France");

上面代碼表明該 Producer 的壓縮算法使用的是 GZIP

有壓縮必有解壓縮，Producer 使用壓縮算法壓縮消息后并發送給服務器后，由 Consumer 消費者進行解壓縮，因為采用的何種壓縮算法是隨著 key、value 一起發送過去的，所以消費者知道采用何種壓縮算法。

Kafka 重要參數配置

本篇文章我們來介紹一下 Kafka 生產者重要的配置，介紹幾個在內存使用、性能和可靠性方面對生產者影響比較大的參數進行說明

key.serializer

用于 key 鍵的序列化，它實現了 org.apache.kafka.common.serialization.Serializer 接口

value.serializer

用于 value 值的序列化，實現了 org.apache.kafka.common.serialization.Serializer 接口

acks

acks 參數指定了要有多少個分區副本接收消息，生產者才認為消息是寫入成功的。此參數對消息丟失的影響較大

• 如果 acks = 0，就表示生產者也不知道自己產生的消息是否被服務器接收了，它才知道它寫成功了。如果發送的途中產生了錯誤，生產者也不知道，它也比較懵逼，因為沒有返回任何消息。這就類似于 UDP 的運輸層協議，只管發，服務器接受不接受它也不關心。
• 如果 acks = 1，只要集群的 Leader 接收到消息，就會給生產者返回一條消息，告訴它寫入成功。如果發送途中造成了網絡異常或者 Leader 還沒選舉出來等其他情況導致消息寫入失敗，生產者會受到錯誤消息，這時候生產者往往會再次重發數據。因為消息的發送也分為 同步 和 異步，Kafka 為了保證消息的高效傳輸會決定是同步發送還是異步發送。如果讓客戶端等待服務器的響應（通過調用 Future 中的 get() 方法），顯然會增加延遲，如果客戶端使用回調，就會解決這個問題。
• 如果 acks = all，這種情況下是只有當所有參與復制的節點都收到消息時，生產者才會接收到一個來自服務器的消息。不過，它的延遲比 acks =1 時更高，因為我們要等待不只一個服務器節點接收消息。

buffer.memory

此參數用來設置生產者內存緩沖區的大小，生產者用它緩沖要發送到服務器的消息。如果應用程序發送消息的速度超過發送到服務器的速度，會導致生產者空間不足。這個時候，send() 方法調用要么被阻塞，要么拋出異常，具體取決于 block.on.buffer.null 參數的設置。

compression.type

此參數來表示生產者啟用何種壓縮算法，默認情況下，消息發送時不會被壓縮。該參數可以設置為 snappy、gzip 和 lz4，它指定了消息發送給 broker 之前使用哪一種壓縮算法進行壓縮。下面是各壓縮算法的對比

retries

生產者從服務器收到的錯誤有可能是臨時性的錯誤（比如分區找不到首領），在這種情況下，reteis 參數的值決定了生產者可以重發的消息次數，如果達到這個次數，生產者會放棄重試并返回錯誤。默認情況下，生產者在每次重試之間等待 100ms，這個等待參數可以通過 retry.backoff.ms 進行修改。

batch.size

當有多個消息需要被發送到同一個分區時，生產者會把它們放在同一個批次里。該參數指定了一個批次可以使用的內存大小，按照字節數計算。當批次被填滿，批次里的所有消息會被發送出去。不過生產者井不一定都會等到批次被填滿才發送，任意條數的消息都可能被發送。

client.id

此參數可以是任意的字符串，服務器會用它來識別消息的來源，一般配置在日志里

max.in.flight.requests.per.connection

此參數指定了生產者在收到服務器響應之前可以發送多少消息，它的值越高，就會占用越多的內存，不過也會提高吞吐量。把它設為1 可以保證消息是按照發送的順序寫入服務器。

timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms

request.timeout.ms 指定了生產者在發送數據時等待服務器返回的響應時間，metadata.fetch.timeout.ms 指定了生產者在獲取元數據（比如目標分區的首領是誰）時等待服務器返回響應的時間。如果等待時間超時，生產者要么重試發送數據，要么返回一個錯誤。timeout.ms 指定了 broker 等待同步副本返回消息確認的時間，與 asks 的配置相匹配—-如果在指定時間內沒有收到同步副本的確認，那么 broker 就會返回一個錯誤。

max.block.ms

此參數指定了在調用 send() 方法或使用 partitionFor() 方法獲取元數據時生產者的阻塞時間當生產者的發送緩沖區已捕，或者沒有可用的元數據時，這些方法就會阻塞。在阻塞時間達到 max.block.ms 時，生產者會拋出超時異常。

max.request.size

該參數用于控制生產者發送的請求大小。它可以指能發送的單個消息的最大值，也可以指單個請求里所有消息的總大小。

receive.buffer.bytes 和 send.buffer.bytes

Kafka 是基于 TCP 實現的，為了保證可靠的消息傳輸，這兩個參數分別指定了 TCP Socket 接收和發送數據包的緩沖區的大小。如果它們被設置為 -1，就使用操作系統的默認值。如果生產者或消費者與 broker 處于不同的數據中心，那么可以適當增大這些值。

Kafka Consumer

應用程序使用 KafkaConsumer 從 Kafka 中訂閱主題并接收來自這些主題的消息，然后再把他們保存起來。應用程序首先需要創建一個 KafkaConsumer 對象，訂閱主題并開始接受消息，驗證消息并保存結果。一段時間后，生產者往主題寫入的速度超過了應用程序驗證數據的速度，這時候該如何處理？如果只使用單個消費者的話，應用程序會跟不上消息生成的速度，就像多個生產者像相同的主題寫入消息一樣，這時候就需要多個消費者共同參與消費主題中的消息，對消息進行分流處理。

Kafka 消費者從屬于消費者群組。一個群組中的消費者訂閱的都是相同的主題，每個消費者接收主題一部分分區的消息。下面是一個 Kafka 分區消費示意圖

上圖中的主題 T1 有四個分區，分別是分區0、分區1、分區2、分區3，我們創建一個消費者群組1，消費者群組中只有一個消費者，它訂閱主題T1，接收到 T1 中的全部消息。由于一個消費者處理四個生產者發送到分區的消息，壓力有些大，需要幫手來幫忙分擔任務，于是就演變為下圖

這樣一來，消費者的消費能力就大大提高了，但是在某些環境下比如用戶產生消息特別多的時候，生產者產生的消息仍舊讓消費者吃不消，那就繼續增加消費者。

如上圖所示，每個分區所產生的消息能夠被每個消費者群組中的消費者消費，如果向消費者群組中增加更多的消費者，那么多余的消費者將會閑置，如下圖所示

向群組中增加消費者是橫向伸縮消費能力的主要方式。總而言之，我們可以通過增加消費組的消費者來進行水平擴展提升消費能力。這也是為什么建議創建主題時使用比較多的分區數，這樣可以在消費負載高的情況下增加消費者來提升性能。另外，消費者的數量不應該比分區數多，因為多出來的消費者是空閑的，沒有任何幫助。

Kafka 一個很重要的特性就是，只需寫入一次消息，可以支持任意多的應用讀取這個消息。換句話說，每個應用都可以讀到全量的消息。為了使得每個應用都能讀到全量消息，應用需要有不同的消費組。對于上面的例子，假如我們新增了一個新的消費組 G2，而這個消費組有兩個消費者，那么就演變為下圖這樣

在這個場景中，消費組 G1 和消費組 G2 都能收到 T1 主題的全量消息，在邏輯意義上來說它們屬于不同的應用。

總結起來就是如果應用需要讀取全量消息，那么請為該應用設置一個消費組；如果該應用消費能力不足，那么可以考慮在這個消費組里增加消費者。

消費者組和分區重平衡

消費者組是什么

消費者組（Consumer Group）是由一個或多個消費者實例（Consumer Instance）組成的群組，具有可擴展性和可容錯性的一種機制。消費者組內的消費者共享一個消費者組ID，這個ID 也叫做 Group ID，組內的消費者共同對一個主題進行訂閱和消費，同一個組中的消費者只能消費一個分區的消息，多余的消費者會閑置，派不上用場。

我們在上面提到了兩種消費方式

• 一個消費者群組消費一個主題中的消息，這種消費模式又稱為點對點的消費方式，點對點的消費方式又被稱為消息隊列
• 一個主題中的消息被多個消費者群組共同消費，這種消費模式又稱為發布-訂閱模式

消費者重平衡

我們從上面的消費者演變圖中可以知道這么一個過程：最初是一個消費者訂閱一個主題并消費其全部分區的消息，后來有一個消費者加入群組，隨后又有更多的消費者加入群組，而新加入的消費者實例分攤了最初消費者的部分消息，這種把分區的所有權通過一個消費者轉到其他消費者的行為稱為重平衡，英文名也叫做 Rebalance 。如下圖所示

重平衡非常重要，它為消費者群組帶來了高可用性 和 伸縮性，我們可以放心的添加消費者或移除消費者，不過在正常情況下我們并不希望發生這樣的行為。在重平衡期間，消費者無法讀取消息，造成整個消費者組在重平衡的期間都不可用。另外，當分區被重新分配給另一個消費者時，消息當前的讀取狀態會丟失，它有可能還需要去刷新緩存，在它重新恢復狀態之前會拖慢應用程序。

消費者通過向組織協調者（Kafka Broker）發送心跳來維護自己是消費者組的一員并確認其擁有的分區。對于不同不的消費群體來說，其組織協調者可以是不同的。只要消費者定期發送心跳，就會認為消費者是存活的并處理其分區中的消息。當消費者檢索記錄或者提交它所消費的記錄時就會發送心跳。

如果過了一段時間 Kafka 停止發送心跳了，會話（Session）就會過期，組織協調者就會認為這個 Consumer 已經死亡，就會觸發一次重平衡。如果消費者宕機并且停止發送消息，組織協調者會等待幾秒鐘，確認它死亡了才會觸發重平衡。在這段時間里，死亡的消費者將不處理任何消息。在清理消費者時，消費者將通知協調者它要離開群組，組織協調者會觸發一次重平衡，盡量降低處理停頓。

重平衡是一把雙刃劍，它為消費者群組帶來高可用性和伸縮性的同時，還有有一些明顯的缺點(bug)，而這些 bug 到現在社區還無法修改。

重平衡的過程對消費者組有極大的影響。因為每次重平衡過程中都會導致萬物靜止，參考 JVM 中的垃圾回收機制，也就是 Stop The World ，STW，(引用自《深入理解 Java 虛擬機》中 p76 關于 Serial 收集器的描述)：

更重要的是它在進行垃圾收集時，必須暫停其他所有的工作線程。直到它收集結束。Stop The World 這個名字聽起來很帥，但這項工作實際上是由虛擬機在后臺自動發起并完成的，在用戶不可見的情況下把用戶正常工作的線程全部停掉，這對很多應用來說都是難以接受的。

也就是說，在重平衡期間，消費者組中的消費者實例都會停止消費，等待重平衡的完成。而且重平衡這個過程很慢……

創建消費者

上面的理論說的有點多，下面就通過代碼來講解一下消費者是如何消費的

在讀取消息之前，需要先創建一個 KafkaConsumer 對象。創建 KafkaConsumer 對象與創建 KafkaProducer 對象十分相似 — 把需要傳遞給消費者的屬性放在 properties 對象中，后面我們會著重討論 Kafka 的一些配置，這里我們先簡單的創建一下，使用3個屬性就足矣，分別是 bootstrap.server，key.deserializer，value.deserializer 。

這三個屬性我們已經用過很多次了，如果你還不是很清楚的話，可以參考 帶你漲姿勢是認識一下Kafka Producer

還有一個屬性是 group.id 這個屬性不是必須的，它指定了 KafkaConsumer 是屬于哪個消費者群組。創建不屬于任何一個群組的消費者也是可以的

Properties properties = new Properties();properties.put("bootstrap.server","192.168.1.9:9092"); properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer"); properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

主題訂閱

創建好消費者之后，下一步就開始訂閱主題了。subscribe() 方法接受一個主題列表作為參數，使用起來比較簡單

consumer.subscribe(Collections.singletonList("customerTopic"));

為了簡單我們只訂閱了一個主題 customerTopic，參數傳入的是一個正則表達式，正則表達式可以匹配多個主題，如果有人創建了新的主題，并且主題的名字與正則表達式相匹配，那么會立即觸發一次重平衡，消費者就可以讀取新的主題。

要訂閱所有與 test 相關的主題，可以這樣做

consumer.subscribe("test.*");

輪詢

我們知道，Kafka 是支持訂閱/發布模式的，生產者發送數據給 Kafka Broker，那么消費者是如何知道生產者發送了數據呢？其實生產者產生的數據消費者是不知道的，KafkaConsumer 采用輪詢的方式定期去 Kafka Broker 中進行數據的檢索，如果有數據就用來消費，如果沒有就再繼續輪詢等待，下面是輪詢等待的具體實現

try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));for (ConsumerRecord<String, String> record : records) {int updateCount = 1;if (map.containsKey(record.value())) {updateCount = (int) map.get(record.value() + 1);}map.put(record.value(), updateCount);}} }finally {consumer.close(); }

• 這是一個無限循環。消費者實際上是一個長期運行的應用程序，它通過輪詢的方式向 Kafka 請求數據。
• 第三行代碼非常重要，Kafka 必須定期循環請求數據，否則就會認為該 Consumer 已經掛了，會觸發重平衡，它的分區會移交給群組中的其它消費者。傳給 poll() 方法的是一個超市時間，用 java.time.Duration 類來表示，如果該參數被設置為 0 ，poll() 方法會立刻返回，否則就會在指定的毫秒數內一直等待 broker 返回數據。
• poll() 方法會返回一個記錄列表。每條記錄都包含了記錄所屬主題的信息，記錄所在分區的信息、記錄在分區中的偏移量，以及記錄的鍵值對。我們一般會遍歷這個列表，逐條處理每條記錄。
• 在退出應用程序之前使用 close() 方法關閉消費者。網絡連接和 socket 也會隨之關閉，并立即觸發一次重平衡，而不是等待群組協調器發現它不再發送心跳并認定它已經死亡。

線程安全性：
在同一個群組中，我們無法讓一個線程運行多個消費者，也無法讓多個線程安全的共享一個消費者。按照規則，一個消費者使用一個線程，如果一個消費者群組中多個消費者都想要運行的話，那么必須讓每個消費者在自己的線程中運行，可以使用 Java 中的 ExecutorService 啟動多個消費者進行進行處理。

消費者配置

到目前為止，我們學習了如何使用消費者 API，不過只介紹了幾個最基本的屬性，Kafka 文檔列出了所有與消費者相關的配置說明。大部分參數都有合理的默認值，一般不需要修改它們，下面我們就來介紹一下這些參數。

• fetch.min.bytes
  該屬性指定了消費者從服務器獲取記錄的最小字節數。broker 在收到消費者的數據請求時，如果可用的數據量小于 fetch.min.bytes 指定的大小，那么它會等到有足夠的可用數據時才把它返回給消費者。這樣可以降低消費者和 broker 的工作負載，因為它們在主題使用頻率不是很高的時候就不用來回處理消息。如果沒有很多可用數據，但消費者的 CPU 使用率很高，那么就需要把該屬性的值設得比默認值大。如果消費者的數量比較多，把該屬性的值調大可以降低 broker 的工作負載。

• fetch.max.wait.ms
  我們通過上面的 fetch.min.bytes 告訴 Kafka，等到有足夠的數據時才會把它返回給消費者。而 fetch.max.wait.ms 則用于指定 broker 的等待時間，默認是 500 毫秒。如果沒有足夠的數據流入 kafka 的話，消費者獲取的最小數據量要求就得不到滿足，最終導致 500 毫秒的延遲。如果要降低潛在的延遲，就可以把參數值設置的小一些。如果 fetch.max.wait.ms 被設置為 100 毫秒的延遲，而 fetch.min.bytes 的值設置為 1MB，那么 Kafka 在收到消費者請求后，要么返回 1MB 的數據，要么在 100 ms 后返回所有可用的數據。就看哪個條件首先被滿足。

• max.partition.fetch.bytes
  該屬性指定了服務器從每個分區里返回給消費者的最大字節數。它的默認值時 1MB，也就是說，KafkaConsumer.poll() 方法從每個分區里返回的記錄最多不超過 max.partition.fetch.bytes 指定的字節。如果一個主題有20個分區和5個消費者，那么每個消費者需要至少4 MB的可用內存來接收記錄。在為消費者分配內存時，可以給它們多分配一些，因為如果群組里有消費者發生崩潰，剩下的消費者需要處理更多的分區。max.partition.fetch.bytes 的值必須比 broker 能夠接收的最大消息的字節數(通過 max.message.size 屬性配置大)，否則消費者可能無法讀取這些消息，導致消費者一直掛起重試。 在設置該屬性時，另外一個考量的因素是消費者處理數據的時間。消費者需要頻繁的調用 poll() 方法來避免會話過期和發生分區再平衡，如果單次調用poll() 返回的數據太多，消費者需要更多的時間進行處理，可能無法及時進行下一個輪詢來避免會話過期。如果出現這種情況，可以把 max.partition.fetch.bytes 值改小，或者延長會話過期時間。

• session.timeout.ms
  這個屬性指定了消費者在被認為死亡之前可以與服務器斷開連接的時間，默認是 3s。如果消費者沒有在 session.timeout.ms 指定的時間內發送心跳給群組協調器，就會被認定為死亡，協調器就會觸發重平衡。把它的分區分配給消費者群組中的其它消費者，此屬性與 heartbeat.interval.ms 緊密相關。heartbeat.interval.ms 指定了 poll() 方法向群組協調器發送心跳的頻率，session.timeout.ms 則指定了消費者可以多久不發送心跳。所以，這兩個屬性一般需要同時修改，heartbeat.interval.ms 必須比 session.timeout.ms 小，一般是 session.timeout.ms 的三分之一。如果 session.timeout.ms 是 3s，那么 heartbeat.interval.ms 應該是 1s。把 session.timeout.ms 值設置的比默認值小，可以更快地檢測和恢復崩憤的節點，不過長時間的輪詢或垃圾收集可能導致非預期的重平衡。把該屬性的值設置得大一些，可以減少意外的重平衡，不過檢測節點崩潰需要更長的時間。

• auto.offset.reset
  該屬性指定了消費者在讀取一個沒有偏移量的分區或者偏移量無效的情況下的該如何處理。它的默認值是 latest，意思指的是，在偏移量無效的情況下，消費者將從最新的記錄開始讀取數據。另一個值是 earliest，意思指的是在偏移量無效的情況下，消費者將從起始位置處開始讀取分區的記錄。

• enable.auto.commit
  我們稍后將介紹幾種不同的提交偏移量的方式。該屬性指定了消費者是否自動提交偏移量，默認值是 true，為了盡量避免出現重復數據和數據丟失，可以把它設置為 false，由自己控制何時提交偏移量。如果把它設置為 true，還可以通過 auto.commit.interval.ms 屬性來控制提交的頻率

• partition.assignment.strategy
  我們知道，分區會分配給群組中的消費者。PartitionAssignor 會根據給定的消費者和主題，決定哪些分區應該被分配給哪個消費者，Kafka 有兩個默認的分配策略Range 和 RoundRobin

• client.id
  該屬性可以是任意字符串，broker 用他來標識從客戶端發送過來的消息，通常被用在日志、度量指標和配額中

• max.poll.records
  該屬性用于控制單次調用 call() 方法能夠返回的記錄數量，可以幫你控制在輪詢中需要處理的數據量。

• receive.buffer.bytes 和 send.buffer.bytes
  socket 在讀寫數據時用到的 TCP 緩沖區也可以設置大小。如果它們被設置為 -1，就使用操作系統默認值。如果生產者或消費者與 broker 處于不同的數據中心內，可以適當增大這些值，因為跨數據中心的網絡一般都有比較高的延遲和比較低的帶寬。

提交和偏移量的概念

特殊偏移

我們上面提到，消費者在每次調用poll() 方法進行定時輪詢的時候，會返回由生產者寫入 Kafka 但是還沒有被消費者消費的記錄，因此我們可以追蹤到哪些記錄是被群組里的哪個消費者讀取的。消費者可以使用 Kafka 來追蹤消息在分區中的位置（偏移量）

消費者會向一個叫做 _consumer_offset 的特殊主題中發送消息，這個主題會保存每次所發送消息中的分區偏移量，這個主題的主要作用就是消費者觸發重平衡后記錄偏移使用的，消費者每次向這個主題發送消息，正常情況下不觸發重平衡，這個主題是不起作用的，當觸發重平衡后，消費者停止工作，每個消費者可能會分到對應的分區，這個主題就是讓消費者能夠繼續處理消息所設置的。

如果提交的偏移量小于客戶端最后一次處理的偏移量，那么位于兩個偏移量之間的消息就會被重復處理

如果提交的偏移量大于最后一次消費時的偏移量，那么處于兩個偏移量中間的消息將會丟失

既然_consumer_offset 如此重要，那么它的提交方式是怎樣的呢？下面我們就來說一下提交方式

KafkaConsumer API 提供了多種方式來提交偏移量

自動提交

最簡單的方式就是讓消費者自動提交偏移量。如果 enable.auto.commit 被設置為true，那么每過 5s，消費者會自動把從 poll() 方法輪詢到的最大偏移量提交上去。提交時間間隔由 auto.commit.interval.ms 控制，默認是 5s。與消費者里的其他東西一樣，自動提交也是在輪詢中進行的。消費者在每次輪詢中會檢查是否提交該偏移量了，如果是，那么就會提交從上一次輪詢中返回的偏移量。

提交當前偏移量

把 auto.commit.offset 設置為 false，可以讓應用程序決定何時提交偏移量。使用 commitSync() 提交偏移量。這個 API 會提交由 poll() 方法返回的最新偏移量，提交成功后馬上返回，如果提交失敗就拋出異常。

commitSync() 將會提交由 poll() 返回的最新偏移量，如果處理完所有記錄后要確保調用了 commitSync()，否則還是會有丟失消息的風險，如果發生了在均衡，從最近一批消息到發生在均衡之間的所有消息都將被重復處理。

異步提交

異步提交 commitAsync() 與同步提交 commitSync() 最大的區別在于異步提交不會進行重試，同步提交會一致進行重試。

同步和異步組合提交

一般情況下，針對偶爾出現的提交失敗，不進行重試不會有太大的問題，因為如果提交失敗是因為臨時問題導致的，那么后續的提交總會有成功的。但是如果在關閉消費者或再均衡前的最后一次提交，就要確保提交成功。

因此，在消費者關閉之前一般會組合使用commitAsync和commitSync提交偏移量。

提交特定的偏移量

消費者API允許調用 commitSync() 和 commitAsync() 方法時傳入希望提交的 partition 和 offset 的 map，即提交特定的偏移量。

總結

以上是生活随笔為你收集整理的Kafka生产者与消费者详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。