kafka 三種消費模式

2021-10-04 14:37:27 字數 1232 閱讀 6159

自動提交offset

以下例項**展示了如何自動提交topic的offset:

public void autooffsetcommit() 

}}

properties的例項props中存放的key意義:

1)bootstrap.servers表示要連線的kafka集群中的節點,其中9092表示埠號;

2)group.id表示kafka消費者組的唯一標識;

2)enable.auto.commit為true,表示在auto.commit.interval.ms時間後會自動提交topic的offset,其中auto.commit.interval.ms預設值為5000ms;

3)其中name1和name2為要消費的topic名稱,由group.id為binghe作為consumer group統一進行管理;

4)key.deserializer和value.deserializer表示指定將位元組序列化為物件。

手動提交offset

生產環境中,需要在資料消費完全後再提交offset,也就是說在資料從kafka的topic取出來後並被邏輯處理後,才算是資料被消費掉,此時需要手動去提交topic的offset。

以下例項**展示了如何手動提交topic的offset:

public void manualoffsetcommit() 

if (buffer.size() >= minbatchsize)

}}

本方案的缺點是必須保證所有資料被處理後,才提交topic的offset。為避免資料的重複消費,可以用第三種方案,根據每個partition的資料消費情況進行提交。

手動提交partition的offset

以下例項**展示了手動提交topic中每乙個partition的offset:

public void manualoffsetcommitofpartition() 

long lastoffset = partitionrecords.get(partitionrecords.size() - 1).offset();

consumer.commitsync(collections.singletonmap(partition, new offsetandmetadata(lastoffset + 1)));}}

} finally

}

Kafka之 三種消費模式

以下例項 展示了如何自動提交topic的offset public void autooffsetcommit properties的例項props中存放的key意義 1 bootstrap.servers表示要連線的kafka集群中的節點,其中9092表示埠號 2 group.id表示kafka消...

kafka消費者的三種模式

採用預設配置情況下,既不能完全保證at least once 也不能完全保證at most once 比如 在自動提交之後,資料消費流程失敗,這樣就會有丟失,不能保證at least once 資料消費成功,但是自動提交失敗,可能會導致重複消費,這樣也不能保證at most once 但是將自動提交...

Kafka的三種ACK機制

kafka producer有三種ack機制 初始化producer時在config中進行配置 0 意味著producer不等待broker同步完成的確認,繼續傳送下一條 批 資訊 提供了最低的延遲。但是最弱的永續性,當伺服器發生故障時,就很可能發生資料丟失。例如leader已經死亡,produce...