如何分析Kafka Consumer

這期內容當中小編將會給大家帶來有關如何分析Kafka Consumer，文章內容豐富且以專業的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

Kafka Consumer消費以組的方式劃分，Topic中的每一個分區只會分給同一個組中的其中一個實例。這是基于隊列模式，如果想基于發布訂閱模式，那訂閱同一個Topic的實例需要指定不同的組名。

必需參數

bootstrap.servers

Kafka服務器

group.id

Consumer Group的名字，唯一標識一個consumer group

key.deserializer

Key的反序列化，二進制的消息Key轉換成具體的類型

value.desrializer

Value的反序列化，二進制的消息內容轉換成具體的類型

主要參數

session.timeout.ms

coordinator檢測失敗的時間，通常需要設置一個較小的值，這樣可以快速檢測到consumer崩潰的情況，盡快開啟rebalance。

max.poll.interval.ms

用于設置消息處理邏輯的最大時間

auto.offset.reset

consumer group無位移信息和位移越界時Kafka對應的策略。consumer group重啟不會使用該策略，因為Kafka已經記錄了group的唯一信息

• earliest:從最早的位移開始消費，不一定就是0

• latest:從最新位移處開始消費

• none:如果無位移信息和位移越界，拋出異常。

enable.auto.commit

指定consumer是否自動提交位移，默認為true

fetch.max.bytes

指定consumer單次獲取數據的最大字節數

max.poll.records

控制poll方法返回的最大消息數量

heartbeat.interval.ms

控制consumer group中成員感知rebalance的時間。

connections.max.idle.ms

空閑連接空閑時間超過該參數，會被關閉。

auto.commit.interval.ms

后臺自動提交位移的時間間隔

消息輪詢Poll

新版Consumer采用了類似Linux I/O模型Poll，使用一個線程管理多個socket連接，然后循環Poll消息。

poll方法返回的條件是要不獲得了足夠多的數據，或者超過了指定的超時時間。

位移管理

新版本的consumer位移已交由內部topic管理(_consumeroffsets)，該Topic有多個分區，每個分區有多個副本(可以通過參數控制)。該內部Topic存在的唯一目的保存consumer提交的位移。

手動提交位移支持同步和異步，提交需要位移需要指定一個Map，key是TopicPartition，value是OffsetAndMetadata，里面存儲了下一條待消費消息的offset。

上述就是小編為大家分享的如何分析Kafka Consumer了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業資訊頻道。