1. 自動(dòng)提交
最簡(jiǎn)單的提交方式是讓消費(fèi)者自動(dòng)提交偏移量约炎。如果 enable.auto.commit 被設(shè)為 true，那么每過(guò) 5s雕旨，消費(fèi)者會(huì)自動(dòng)把從 poll() 方法接收到的最大偏移量提交上去喳魏。提交時(shí)間間隔由 auto.commit.interval.ms 控制，默認(rèn)值是5s揖曾。消費(fèi)者每次獲取新數(shù)據(jù)時(shí)都會(huì)先把上一次poll()方法返回的最大偏移量提交上去。
可能造成的問(wèn)題：數(shù)據(jù)重復(fù)讀
假設(shè)我們?nèi)匀皇褂媚J(rèn)的 5s 提交時(shí)間間隔亥啦，在最近一次提交之后的 3s 發(fā)生了再均衡炭剪，再均衡之后，消費(fèi)者從最后一次提交的偏移量位置開始讀取消息翔脱。這個(gè)時(shí)候偏移量已經(jīng)落后了 3s念祭，所以在這 3s內(nèi)到達(dá)的消息會(huì)被重復(fù)處理“欤可以通過(guò)修改提交時(shí)間間隔來(lái)更頻繁地提交偏移量，減小可能出現(xiàn)重復(fù)消息的時(shí)間窗隶糕，不過(guò)這種情況是無(wú)法完全避免的瓷产。
2. 手動(dòng)提交
(1) 同步提交
// 把a(bǔ)uto.commit.offset設(shè)為false，讓應(yīng)用程序決定何時(shí)提交偏移量
props.put("auto.commit.offset", false);
try{
while(true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for(ConsumerRecord<String, String> record : records) {
// 假設(shè)把記錄內(nèi)容打印出來(lái)就算處理完畢
System.out.println("value = " + record.value() + ", topic = " + record.topic() +
", partition = " + record.partition() + ", offset = " + record.offset());
}
try{
// 只要沒(méi)有發(fā)生不可恢復(fù)的錯(cuò)誤枚驻，commitSync() 方法會(huì)一直嘗試直至提交成功
// 如果提交失敗濒旦，我們也只能把異常記錄到錯(cuò)誤日志里
consumer.commitSync();
}catch(CommitFailedException e) {
System.err.println("commit failed!" + e.getMessage());
}
}
}finally {
consumer.close();
}
(2) 異步提交
手動(dòng)提交有一個(gè)不足之處，在 broker 對(duì)提交請(qǐng)求作出回應(yīng)之前再登，應(yīng)用程序會(huì)一直阻塞尔邓，這樣會(huì)限制應(yīng)用程序的吞吐量。我們可以通過(guò)降低提交頻率來(lái)提升吞吐量锉矢，但如果發(fā)生了再均衡梯嗽，會(huì)增加重復(fù)消息的數(shù)量。
這個(gè)時(shí)候可以使用異步提交沽损，只管發(fā)送提交請(qǐng)求灯节，無(wú)需等待 broker 的響應(yīng)。
// 把a(bǔ)uto.commit.offset設(shè)為false，讓應(yīng)用程序決定何時(shí)提交偏移量
props.put("auto.commit.offset", false);
try{
while(true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for(ConsumerRecord<String, String> record : records) {
System.out.println("value = " + record.value() + ", topic = " + record.topic() +
", partition = " + record.partition() + ", offset = " + record.offset());
}
// 提交最后一個(gè)偏移量炎疆，然后繼續(xù)做其他事情卡骂。
consumer.commitAsync();
}
}finally {
consumer.close();
}
在成功提交或碰到無(wú)法恢復(fù)的錯(cuò)誤之前，commitSync()會(huì)一直重試形入，但是commitAsync()不會(huì)全跨，這也是commitAsync()不好的一個(gè)地方。它之所以不進(jìn)行重試亿遂，是因?yàn)樵谒盏椒?wù)器響應(yīng)的時(shí)候浓若，可能有一個(gè)更大的偏移量已經(jīng)提交成功。
假設(shè)我們發(fā)出一個(gè)請(qǐng)求用于提交偏移量2000崩掘，這個(gè)時(shí)候發(fā)生了短暫的通信問(wèn)題七嫌，服務(wù)器收不到請(qǐng)求，自然也不會(huì)作出任何響應(yīng)苞慢。與此同時(shí)诵原，我們處理了另外一批消息，并成功提交了偏移量3000挽放。如果commitAsync()重新嘗試提交偏移量2000绍赛，它有可能在偏移量3000之后提交成功。這個(gè)時(shí)候如果發(fā)生再均衡辑畦，就會(huì)出現(xiàn)重復(fù)消息吗蚌。
commitAsync()也支持回調(diào)，在broker作出響應(yīng)時(shí)會(huì)執(zhí)行回調(diào)：
// 把a(bǔ)uto.commit.offset設(shè)為false纯出，讓應(yīng)用程序決定何時(shí)提交偏移量
props.put("auto.commit.offset", false);
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> 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<TopicPartition, OffsetAndMetadata> 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();
}
可以在回調(diào)中重試失敗的提交蚯妇，以下為思路：
使用一個(gè)單調(diào)遞增的序列號(hào)來(lái)維護(hù)異步提交的順序。在每次提交偏移量之后或在回調(diào)里提交偏移量時(shí)遞增序列號(hào)暂筝。在進(jìn)行重試前箩言，先檢查回調(diào)的序列號(hào)和即將提交的偏移量是否相等，如果相等焕襟，說(shuō)明沒(méi)有新的提交陨收，那么可以安全地進(jìn)行重試。如果序列號(hào)比較大鸵赖，說(shuō)明有一個(gè)新的提交已經(jīng)發(fā)送出去了务漩，應(yīng)該停止重試。
(3) 同步和異步組合提交
一般情況下它褪，針對(duì)偶爾出現(xiàn)的提交失敗饵骨，不進(jìn)行重試不會(huì)有太大問(wèn)題，因?yàn)槿绻峤皇∈且驗(yàn)榕R時(shí)問(wèn)題導(dǎo)致的茫打，那么后續(xù)的提交總會(huì)有成功的宏悦。但如果這是發(fā)生在關(guān)閉消費(fèi)者或再均衡前的最后一次提交镐确，就要確保能夠提交成功。
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
+ record.partition() + ", offset = " + record.offset());
}
// 如果一切正常饼煞，我們使用 commitAsync() 方法來(lái)提交
// 這樣速度更快源葫，而且即使這次提交失敗，下一次提交很可能會(huì)成功
consumer.commitAsync();
}
}catch (Exception e) {
e.printStackTrace();
}finally {
try {
// 使用 commitSync() 方法會(huì)一直重試砖瞧，直到提交成功或發(fā)生無(wú)法恢復(fù)的錯(cuò)誤
// 確保關(guān)閉消費(fèi)者之前成功提交了偏移量
consumer.commitSync();
}finally {
consumer.close();
}
}
(4) 提交特定的偏移量
不管是自動(dòng)提交還是使用commitAsync()或者commitSync()來(lái)提交偏移量息堂，提交的都是 poll() 方法返回的那批數(shù)據(jù)的最大偏移量，想要自定義在什么時(shí)候提交偏移量可以這么做：
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
int count = 0;
......
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> 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) {
// 這里調(diào)用的是 commitAsync()块促，不過(guò)調(diào)用 commitSync() 也是完全可以的
// 當(dāng)然荣堰，在提交特定偏移量時(shí)，仍然要處理可能發(fā)生的錯(cuò)誤
consumer.commitAsync(currentOffsets, null);
}
count++;
}
}
}finally {
consumer.close();
}
3. 分區(qū)再均衡監(jiān)聽器
消費(fèi)者在退出和進(jìn)行分區(qū)再均衡之前竭翠，應(yīng)該做一些正確的事情：
- 提交最后一個(gè)已處理記錄的偏移量(必須做)
- 根據(jù)之前處理數(shù)據(jù)的業(yè)務(wù)不同振坚，你可能還需要關(guān)閉數(shù)據(jù)庫(kù)連接池、清空緩存等
程序如何能得知集群要進(jìn)行"分區(qū)再均衡"了斋扰？消費(fèi)者 API 提供了再均衡監(jiān)聽器渡八，以下程序可以做到 kafka 消費(fèi)數(shù)據(jù)的 Exactly Once 語(yǔ)義：
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 再均衡監(jiān)聽器
* @Author YangYunhe
* @Date 2018-06-27 17:35:05
*/
public class RebalanceListenerConsumer {
public static void main(String[] args) {
Map<TopicPartition, OffsetAndMetadata> 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");
// 把a(bǔ)uto.commit.offset設(shè)為false，讓應(yīng)用程序決定何時(shí)提交偏移量
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<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
/*
* 再均衡開始之前和消費(fèi)者停止讀取消息之后被調(diào)用
* 如果在這里提交偏移量传货，下一個(gè)接管分區(qū)的消費(fèi)者就知道該從哪里開始讀取了
*/
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// 如果發(fā)生再均衡屎鳍，我們要在即將失去分區(qū)所有權(quán)時(shí)提交偏移量
// 要注意，提交的是最近處理過(guò)的偏移量问裕，而不是批次中還在處理的最后一個(gè)偏移量
System.out.println("Lost partitions in rebalance. Committing current offsets:" + currentOffsets);
consumer.commitSync(currentOffsets);
}
/*
* 在重新分配分區(qū)之后和新的消費(fèi)者開始讀取消息之前被調(diào)用
*/
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
long committedOffset = -1;
for(TopicPartition topicPartition : partitions) {
// 獲取該分區(qū)已經(jīng)消費(fèi)的偏移量
committedOffset = consumer.committed(topicPartition).offset();
// 重置偏移量到上一次提交的偏移量的下一個(gè)位置處開始消費(fèi)
consumer.seek(topicPartition, committedOffset + 1);
}
}
});
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> 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!");
}
}
}
}
當(dāng)然你也可以選擇再均衡后從頭開始消費(fèi)：
consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
System.out.println("starting partitions rebalance...");
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
consumer.seekToBeginning(partitions);
}
});
以上代碼與 props.put("auto.offset.reset", "earliest");是等效的逮壁。
設(shè)置從最新消息開始消費(fèi)：
consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
System.out.println("starting partitions rebalance...");
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
consumer.seekToEnd(partitions);
}
});
以上代碼與props.put("auto.offset.reset", "latest");等效。
4. 涉及到數(shù)據(jù)庫(kù)的 Exactly Once 語(yǔ)義的實(shí)現(xiàn)思路
當(dāng)處理 Kafka 中的數(shù)據(jù)涉及到數(shù)據(jù)庫(kù)時(shí)粮宛，那么即使每處理一條數(shù)據(jù)提交一次偏移量窥淆，也可以造成數(shù)據(jù)重復(fù)處理或者丟失數(shù)據(jù)，看以下為偽代碼：
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
......
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1);
// 處理數(shù)據(jù)
processRecord(record);
// 把數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)中
storeRecordInDB(record);
// 提交偏移量
consumer.commitAsync(currentOffsets);
}
}
假設(shè)把數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)后巍杈，沒(méi)有來(lái)得及提交偏移量程序就因某種原因掛掉了忧饭，那么程序再次啟動(dòng)后就會(huì)重復(fù)處理數(shù)據(jù)，數(shù)據(jù)庫(kù)中會(huì)有重復(fù)的數(shù)據(jù)秉氧。
如果把存儲(chǔ)到數(shù)據(jù)庫(kù)和提交偏移量在一個(gè)原子操作里完成，就可以避免這樣的問(wèn)題蜒秤，但數(shù)據(jù)存到數(shù)據(jù)庫(kù)汁咏，偏移量保存到kafka是無(wú)法實(shí)現(xiàn)原子操作的，而如果把數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)中作媚，偏移量也存儲(chǔ)到數(shù)據(jù)庫(kù)中攘滩，這樣就可以利用數(shù)據(jù)庫(kù)的事務(wù)來(lái)把這兩個(gè)操作設(shè)為一個(gè)原子操作，同時(shí)結(jié)合再均衡監(jiān)聽器就可以實(shí)現(xiàn) Exactly Once 語(yǔ)義纸泡，以下為偽代碼：
consumer.subscribe(Collections<String> topics, new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// 發(fā)生分區(qū)再均衡之前漂问，提交事務(wù)
commitDBTransaction();
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
// 再均衡之后，從數(shù)據(jù)庫(kù)獲得消費(fèi)偏移量
for(TopicPartition topicPartition : partitions) {
consumer.seek(topicPartition, getOffsetFromDB(topicPartition));
}
}
});
/**
* 消費(fèi)之前調(diào)用一次 poll()，讓消費(fèi)者加入到消費(fèi)組中蚤假，并獲取分配的分區(qū)
* 然后馬上調(diào)用 seek() 方法定位分區(qū)的偏移量
* seek() 設(shè)置消費(fèi)偏移量栏饮，設(shè)置的偏移量是從數(shù)據(jù)庫(kù)讀出來(lái)的，說(shuō)明本次設(shè)置的偏移量已經(jīng)被處理過(guò)
* 下一次調(diào)用 poll() 就會(huì)在本次設(shè)置的偏移量上加1磷仰，開始處理沒(méi)有處理過(guò)的數(shù)據(jù)
* 如果seek()發(fā)生錯(cuò)誤袍嬉，比如偏移量不存在，則會(huì)拋出異常
*/
consumer.poll(0);
for(TopicPartition topicPartition : consumer.assignment()) {
consumer.seek(topicPartition, getOffsetFromDB(topicPartition));
}
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
// 處理數(shù)據(jù)
processRecord(record);
// 把數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)中
storeRecordInDB(record);
// 把偏移量存儲(chǔ)到數(shù)據(jù)庫(kù)中
storeOffsetInDB(record.topic(), record.partition(), record.offset());
}
// 以上3步為一個(gè)事務(wù)灶平，提交事務(wù)伺通，這里在每個(gè)批次末尾提交一次事務(wù)，是為了提高性能
commitDBTransaction();
}
把偏移量和記錄保存到用一個(gè)外部系統(tǒng)來(lái)實(shí)現(xiàn) Exactly Once 有很多方法逢享，但核心思想都是：結(jié)合 ConsumerRebalanceListener 和 seek() 方法來(lái)確保能夠及時(shí)保存偏移量罐监，并保證消費(fèi)者總是能夠從正確的位置開始讀取消息。
