1. 自動提交

最簡單的提交方式是讓消費者自動提交偏移量。如果 enable.auto.commit 被設為 true，那么每過 5s，消費者會自動把從 poll() 方法接收到的最大偏移量提交上去。提交時間間隔由 auto.commit.interval.ms 控制，默認值是5s。消費者每次獲取新數據時都會先把上一次poll()方法返回的最大偏移量提交上去。

可能造成的問題：數據重復讀

假設我們仍然使用默認的 5s 提交時間間隔，在最近一次提交之后的 3s 發生了再均衡，再均衡之后，消費者從最后一次提交的偏移量位置開始讀取消息。這個時候偏移量已經落后了 3s，所以在這 3s內到達的消息會被重復處理。可以通過修改提交時間間隔來更頻繁地提交偏移量，減小可能出現重復消息的時間窗，不過這種情況是無法完全避免的。

2. 手動提交

(1) 同步提交

// 把auto.commit.offset設為false，讓應用程序決定何時提交偏移量

props.put("auto.commit.offset", false);

try{

while(true) {

ConsumerRecords records = consumer.poll(1000);

for(ConsumerRecord record : records) {

// 假設把記錄內容打印出來就算處理完畢

System.out.println("value = " + record.value() + ", topic = " + record.topic() +

", partition = " + record.partition() + ", offset = " + record.offset());

}

try{

// 只要沒有發生不可恢復的錯誤，commitSync() 方法會一直嘗試直至提交成功

// 如果提交失敗，我們也只能把異常記錄到錯誤日志里

consumer.commitSync();

}catch(CommitFailedException e) {

System.err.println("commit failed!" + e.getMessage());

}

}

}finally {

consumer.close();

}

(2) 異步提交

手動提交有一個不足之處，在 broker 對提交請求作出回應之前，應用程序會一直阻塞，這樣會限制應用程序的吞吐量。我們可以通過降低提交頻率來提升吞吐量，但如果發生了再均衡，會增加重復消息的數量。

這個時候可以使用異步提交，只管發送提交請求，無需等待 broker 的響應。

// 把auto.commit.offset設為false，讓應用程序決定何時提交偏移量

props.put("auto.commit.offset", false);

try{

while(true) {

ConsumerRecords records = consumer.poll(1000);

for(ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() +

", partition = " + record.partition() + ", offset = " + record.offset());

}

// 提交最后一個偏移量，然后繼續做其他事情。

consumer.commitAsync();

}

}finally {

consumer.close();

}

在成功提交或碰到無法恢復的錯誤之前，commitSync()會一直重試，但是commitAsync()不會，這也是commitAsync()不好的一個地方。它之所以不進行重試，是因為在它收到服務器響應的時候，可能有一個更大的偏移量已經提交成功。

假設我們發出一個請求用于提交偏移量2000，這個時候發生了短暫的通信問題，服務器收不到請求，自然也不會作出任何響應。與此同時，我們處理了另外一批消息，并成功提交了偏移量3000。如果commitAsync()重新嘗試提交偏移量2000，它有可能在偏移量3000之后提交成功。這個時候如果發生再均衡，就會出現重復消息。

commitAsync()也支持回調，在broker作出響應時會執行回調：

// 把auto.commit.offset設為false，讓應用程序決定何時提交偏移量

props.put("auto.commit.offset", false);

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

}

consumer.commitAsync(new OffsetCommitCallback() {

@Override

public void onComplete(Map offsets, Exception exception) {

if(offsets != null) {

System.out.println("commit offset successful!");

}

if(exception != null) {

System.out.println("commit offset fail!" + exception.getMessage());

}

}

});

}

} finally {

consumer.close();

}

可以在回調中重試失敗的提交，以下為思路：

使用一個單調遞增的序列號來維護異步提交的順序。在每次提交偏移量之后或在回調里提交偏移量時遞增序列號。在進行重試前，先檢查回調的序列號和即將提交的偏移量是否相等，如果相等，說明沒有新的提交，那么可以安全地進行重試。如果序列號比較大，說明有一個新的提交已經發送出去了，應該停止重試。

(3) 同步和異步組合提交

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

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

}

// 如果一切正常，我們使用 commitAsync() 方法來提交

// 這樣速度更快，而且即使這次提交失敗，下一次提交很可能會成功

consumer.commitAsync();

}

}catch (Exception e) {

e.printStackTrace();

}finally {

try {

// 使用 commitSync() 方法會一直重試，直到提交成功或發生無法恢復的錯誤

// 確保關閉消費者之前成功提交了偏移量

consumer.commitSync();

}finally {

consumer.close();

}

}

(4) 提交特定的偏移量

不管是自動提交還是使用commitAsync()或者commitSync()來提交偏移量，提交的都是 poll() 方法返回的那批數據的最大偏移量，想要自定義在什么時候提交偏移量可以這么做：

Map currentOffsets = new HashMap<>();

int count = 0;

......

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),

new OffsetAndMetadata(record.offset() + 1, "no metadata"));

if (count % 1000 == 0) {

// 這里調用的是 commitAsync()，不過調用 commitSync() 也是完全可以的

// 當然，在提交特定偏移量時，仍然要處理可能發生的錯誤

consumer.commitAsync(currentOffsets, null);

}

count++;

}

}

}finally {

consumer.close();

}

3. 分區再均衡監聽器

消費者在退出和進行分區再均衡之前，應該做一些正確的事情：

提交最后一個已處理記錄的偏移量(必須做)

根據之前處理數據的業務不同，你可能還需要關閉數據庫連接池、清空緩存等

程序如何能得知集群要進行"分區再均衡"了？消費者 API 提供了再均衡監聽器，以下程序可以做到 kafka 消費數據的 Exactly Once 語義：

package com.bonc.rdpe.kafka110.consumer;

import java.util.Collection;

import java.util.Collections;

import java.util.HashMap;

import java.util.Map;

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;

import org.apache.kafka.common.TopicPartition;

/**

* @Title RebalanceListenerConsumer.java

* @Description 再均衡監聽器

* @Author YangYunhe

* @Date 2018-06-27 17:35:05

*/

public class RebalanceListenerConsumer {

public static void main(String[] args) {

Map currentOffsets = new HashMap<>();

Properties props = new Properties();

props.put("bootstrap.servers", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");

// 把auto.commit.offset設為false，讓應用程序決定何時提交偏移量

props.put("auto.commit.offset", false);

props.put("group.id", "dev3-yangyunhe-group001");

props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer consumer = new KafkaConsumer<>(props);

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

/*

* 再均衡開始之前和消費者停止讀取消息之后被調用

* 如果在這里提交偏移量，下一個接管分區的消費者就知道該從哪里開始讀取了

*/

@Override

public void onPartitionsRevoked(Collection partitions) {

// 如果發生再均衡，我們要在即將失去分區所有權時提交偏移量

// 要注意，提交的是最近處理過的偏移量，而不是批次中還在處理的最后一個偏移量

System.out.println("Lost partitions in rebalance. Committing current offsets:" + currentOffsets);

consumer.commitSync(currentOffsets);

}

/*

* 在重新分配分區之后和新的消費者開始讀取消息之前被調用

*/

@Override

public void onPartitionsAssigned(Collection partitions) {

long committedOffset = -1;

for(TopicPartition topicPartition : partitions) {

// 獲取該分區已經消費的偏移量

committedOffset = consumer.committed(topicPartition).offset();

// 重置偏移量到上一次提交的偏移量的下一個位置處開始消費

consumer.seek(topicPartition, committedOffset + 1);

}

}

});

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),

new OffsetAndMetadata(record.offset() + 1, "no metadata"));

}

consumer.commitAsync(currentOffsets, null);

}

} catch (Exception e) {

e.printStackTrace();

} finally {

try{

consumer.commitSync(currentOffsets);

} catch (Exception e) {

e.printStackTrace();

} finally {

consumer.close();

System.out.println("Closed consumer successfully!");

}

}

}

}

當然你也可以選擇再均衡后從頭開始消費：

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

@Override

public void onPartitionsRevoked(Collection partitions) {

System.out.println("starting partitions rebalance...");

}

@Override

public void onPartitionsAssigned(Collection partitions) {

consumer.seekToBeginning(partitions);

}

});

以上代碼與 props.put("auto.offset.reset", "earliest");是等效的。

設置從最新消息開始消費：

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

@Override

public void onPartitionsRevoked(Collection partitions) {

System.out.println("starting partitions rebalance...");

}

@Override

public void onPartitionsAssigned(Collection partitions) {

consumer.seekToEnd(partitions);

}

});

以上代碼與props.put("auto.offset.reset", "latest");等效。

4. 涉及到數據庫的 Exactly Once 語義的實現思路

當處理 Kafka 中的數據涉及到數據庫時，那么即使每處理一條數據提交一次偏移量，也可以造成數據重復處理或者丟失數據，看以下為偽代碼：

Map currentOffsets = new HashMap<>();

......

while (true) {

ConsumerRecords records = consumer.poll(100);

for (ConsumerRecord record : records) {

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),

new OffsetAndMetadata(record.offset() + 1);

// 處理數據

processRecord(record);

// 把數據存儲到數據庫中

storeRecordInDB(record);

// 提交偏移量

consumer.commitAsync(currentOffsets);

}

}

假設把數據存儲到數據庫后，沒有來得及提交偏移量程序就因某種原因掛掉了，那么程序再次啟動后就會重復處理數據，數據庫中會有重復的數據。

如果把存儲到數據庫和提交偏移量在一個原子操作里完成，就可以避免這樣的問題，但數據存到數據庫，偏移量保存到kafka是無法實現原子操作的，而如果把數據存儲到數據庫中，偏移量也存儲到數據庫中，這樣就可以利用數據庫的事務來把這兩個操作設為一個原子操作，同時結合再均衡監聽器就可以實現 Exactly Once 語義，以下為偽代碼：

consumer.subscribe(Collections topics, new ConsumerRebalanceListener() {

@Override

public void onPartitionsRevoked(Collection partitions) {

// 發生分區再均衡之前，提交事務

commitDBTransaction();

}

@Override

public void onPartitionsAssigned(Collection partitions) {

// 再均衡之后，從數據庫獲得消費偏移量

for(TopicPartition topicPartition : partitions) {

consumer.seek(topicPartition, getOffsetFromDB(topicPartition));

}

}

});

/**

* 消費之前調用一次 poll()，讓消費者加入到消費組中，并獲取分配的分區

* 然后馬上調用 seek() 方法定位分區的偏移量

* seek() 設置消費偏移量，設置的偏移量是從數據庫讀出來的，說明本次設置的偏移量已經被處理過

* 下一次調用 poll() 就會在本次設置的偏移量上加1，開始處理沒有處理過的數據

* 如果seek()發生錯誤，比如偏移量不存在，則會拋出異常

*/

consumer.poll(0);

for(TopicPartition topicPartition : consumer.assignment()) {

consumer.seek(topicPartition, getOffsetFromDB(topicPartition));

}

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

// 處理數據

processRecord(record);

// 把數據存儲到數據庫中

storeRecordInDB(record);

// 把偏移量存儲到數據庫中

storeOffsetInDB(record.topic(), record.partition(), record.offset());

}

// 以上3步為一個事務，提交事務，這里在每個批次末尾提交一次事務，是為了提高性能

commitDBTransaction();

}

把偏移量和記錄保存到用一個外部系統來實現 Exactly Once 有很多方法，但核心思想都是：結合 ConsumerRebalanceListener 和 seek() 方法來確保能夠及時保存偏移量，并保證消費者總是能夠從正確的位置開始讀取消息。

總結

以上是生活随笔為你收集整理的kafka偏移量保存到mysql里_Kafka 新版消费者 API（二）：提交偏移量的全部內容，希望文章能夠幫你解決所遇到的問題。

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