如何解決 Kafka 消息重復(fù)問(wèn)題！
閱讀本文大概需要 7 分鐘。
來(lái)自：juejin.cn/post/7172897190627508237
一、前言
整理下消息重復(fù)的幾個(gè)場(chǎng)景：
場(chǎng)景一：
leader分區(qū)不可用了，拋LeaderNotAvailableException異常，等待選出新leader分區(qū)。場(chǎng)景二：
Controller所在Broker掛了，拋NotControllerException異常，等待Controller重新選舉。場(chǎng)景三：網(wǎng)絡(luò)異常、斷網(wǎng)、網(wǎng)絡(luò)分區(qū)、丟包等，拋
NetworkException異常，等待網(wǎng)絡(luò)恢復(fù)。
poll 一批數(shù)據(jù)，處理完畢還沒(méi)提交 offset ，機(jī)子宕機(jī)重啟了，又會(huì) poll 上批數(shù)據(jù)，再度消費(fèi)就造成了消息重復(fù)。怎么解決？
最多一次（
at most once）： 消息只發(fā)一次，消息可能會(huì)丟失，但絕不會(huì)被重復(fù)發(fā)送。例如：mqtt中QoS = 0。至少一次（
at least once）： 消息至少發(fā)一次，消息不會(huì)丟失，但有可能被重復(fù)發(fā)送。例如：mqtt中QoS = 1精確一次（
exactly once）： 消息精確發(fā)一次，消息不會(huì)丟失，也不會(huì)被重復(fù)發(fā)送。例如：mqtt中QoS = 2。
了解了這三種語(yǔ)義，再來(lái)看如何解決消息重復(fù)，即如何實(shí)現(xiàn)精準(zhǔn)一次，可分為三種方法：
Kafka冪等性Producer： 保證生產(chǎn)端發(fā)送消息冪等。局限性，是只能保證單分區(qū)且單會(huì)話（重啟后就算新會(huì)話）Kafka事務(wù)： 保證生產(chǎn)端發(fā)送消息冪等。解決冪等Producer的局限性。消費(fèi)端冪等： 保證消費(fèi)端接收消息冪等。蔸底方案。
1）Kafka 冪等性 Producer
Properties props = new Properties();
props.put("enable.idempotence", ture); // 1. 設(shè)置冪等
props.put("acks", "all"); // 2. 當(dāng) enable.idempotence 為 true，這里默認(rèn)為 all
props.put("max.in.flight.requests.per.connection", 5); // 3. 注意
acks，注意：一定要設(shè)置 acks=all，否則會(huì)拋異常。max.in.flight.requests.per.connection 需要 <= 5，否則會(huì)拋異常 OutOfOrderSequenceException。0.11 >= Kafka < 1.1,max.in.flight.request.per.connection = 1Kafka >= 1.1,max.in.flight.request.per.connection <= 5
為了更好理解，需要了解下Kafka 冪等機(jī)制：
Producer 每次啟動(dòng)后，會(huì)向 Broker 申請(qǐng)一個(gè)全局唯一的 pid。（重啟后 pid 會(huì)變化，這也是弊端之一）Sequence Numbe：針對(duì)每個(gè) <Topic, Partition> 都對(duì)應(yīng)一個(gè)從0開(kāi)始單調(diào)遞增的 Sequence，同時(shí) Broker端會(huì)緩存這個(gè) seq num<pid, seq num> 去 Broker 里對(duì)應(yīng)的隊(duì)列 ProducerStateEntry.Queue（默認(rèn)隊(duì)列長(zhǎng)度為 5）查詢是否存在如果
nextSeq == lastSeq + 1，即服務(wù)端seq + 1 == 生產(chǎn)傳入seq，則接收。如果
nextSeq == 0 && lastSeq == Int.MaxValue，即剛初始化，也接收。反之，要么重復(fù)，要么丟消息，均拒絕。
這種設(shè)計(jì)針對(duì)解決了兩個(gè)問(wèn)題：
「消息重復(fù)：」 場(chǎng)景
Broker保存消息后還沒(méi)發(fā)送ack就宕機(jī)了，這時(shí)候Producer就會(huì)重試，這就造成消息重復(fù)。「消息亂序：」 避免場(chǎng)景，前一條消息發(fā)送失敗而其后一條發(fā)送成功，前一條消息重試后成功，造成的消息亂序。
那什么時(shí)候該使用冪等：
如果已經(jīng)使用
acks=all，使用冪等也可以。如果已經(jīng)使用
acks=0或者acks=1，說(shuō)明你的系統(tǒng)追求高性能，對(duì)數(shù)據(jù)一致性要求不高。不要使用冪等。
2）Kafka 事務(wù)
Kafka 事務(wù)解決冪等的弊端：?jiǎn)螘?huì)話且單分區(qū)冪等。Tips：」 這塊篇幅較長(zhǎng)，這先稍微提及下使用，之后另起一篇。Properties props = new Properties();
props.put("enable.idempotence", ture); // 1. 設(shè)置冪等
props.put("acks", "all"); // 2. 當(dāng) enable.idempotence 為 true，這里默認(rèn)為 all
props.put("max.in.flight.requests.per.connection", 5); // 3. 最大等待數(shù)
props.put("transactional.id", "my-transactional-id"); // 4. 設(shè)定事務(wù) id
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 初始化事務(wù)
producer.initTransactions();
try{
// 開(kāi)始事務(wù)
producer.beginTransaction();
// 發(fā)送數(shù)據(jù)
producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));
// 數(shù)據(jù)發(fā)送及 Offset 發(fā)送均成功的情況下，提交事務(wù)
producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
// 數(shù)據(jù)發(fā)送或者 Offset 發(fā)送出現(xiàn)異常時(shí)，終止事務(wù)
producer.abortTransaction();
} finally {
// 關(guān)閉 Producer 和 Consumer
producer.close();
consumer.close();
}
Consumer 需要設(shè)置下配置：isolation.level 參數(shù)「
read_uncommitted：」 這是默認(rèn)值，表明Consumer能夠讀取到Kafka寫(xiě)入的任何消息，不論事務(wù)型Producer提交事務(wù)還是終止事務(wù)，其寫(xiě)入的消息都可以讀取。如果你用了事務(wù)型Producer，那么對(duì)應(yīng)的Consumer就不要使用這個(gè)值。「
read_committed：」 表明Consumer只會(huì)讀取事務(wù)型Producer成功提交事務(wù)寫(xiě)入的消息。當(dāng)然了，它也能看到非事務(wù)型Producer寫(xiě)入的所有消息。
3）消費(fèi)端冪等
上述栗子中，消費(fèi)端拉取到一條消息后，開(kāi)啟事務(wù)，將消息
Id新增到本地消息表中，同時(shí)更新訂單信息。如果消息重復(fù)，則新增操作
insert會(huì)異常，同時(shí)觸發(fā)事務(wù)回滾。
二、案例：Kafka 冪等性 Producer 使用
Docker 啟動(dòng)$ docker run -d --name zookeeper -p 2181:2181 zookeeper
a86dff3689b68f6af7eb3da5a21c2dba06e9623f3c961154a8bbbe3e9991dea4
2.7.1，源碼編譯啟動(dòng)（看上文源碼搭建啟動(dòng)）Kafka 源碼中 exmaple 中Kafka 提供的腳本# 舉個(gè)栗子：topic 需要自己去修改
$ cd ./kafka-2.7.1-src/bin
$ ./kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic
topic ：1副本，2 分區(qū)$ ./kafka-topics.sh --bootstrap-server localhost:9092 --topic myTopic --create --replication-factor 1 --partitions 2
# 查看
$ ./kafka-topics.sh --bootstrap-server broker:9092 --topic myTopic --describe
public class KafkaProducerApplication {
private final Producer<String, String> producer;
final String outTopic;
public KafkaProducerApplication(final Producer<String, String> producer,
final String topic) {
this.producer = producer;
outTopic = topic;
}
public void produce(final String message) {
final String[] parts = message.split("-");
final String key, value;
if (parts.length > 1) {
key = parts[0];
value = parts[1];
} else {
key = null;
value = parts[0];
}
final ProducerRecord<String, String> producerRecord
= new ProducerRecord<>(outTopic, key, value);
producer.send(producerRecord,
(recordMetadata, e) -> {
if(e != null) {
e.printStackTrace();
} else {
System.out.println("key/value " + key + "/" + value + "\twritten to topic[partition] " + recordMetadata.topic() + "[" + recordMetadata.partition() + "] at offset " + recordMetadata.offset());
}
}
);
}
public void shutdown() {
producer.close();
}
public static void main(String[] args) {
final Properties props = new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.CLIENT_ID_CONFIG, "myApp");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
final String topic = "myTopic";
final Producer<String, String> producer = new KafkaProducer<>(props);
final KafkaProducerApplication producerApp = new KafkaProducerApplication(producer, topic);
String filePath = "/home/donald/Documents/Code/Source/kafka-2.7.1-src/examples/src/main/java/kafka/examples/input.txt";
try {
List<String> linesToProduce = Files.readAllLines(Paths.get(filePath));
linesToProduce.stream().filter(l -> !l.trim().isEmpty())
.forEach(producerApp::produce);
System.out.println("Offsets and timestamps committed in batch from " + filePath);
} catch (IOException e) {
System.err.printf("Error reading file %s due to %s %n", filePath, e);
} finally {
producerApp.shutdown();
}
}
}
$ ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic
修改配置 acks
acks 配置，生產(chǎn)者啟動(dòng)后結(jié)果是怎樣的：修改配置
acks = 1修改配置
acks = 0
Exception in thread "main" org.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer.
Otherwise we cannot guarantee idempotence.
修改配置 max.in.flight.requests.per.connection
max.in.flight.requests.per.connection > 5 會(huì)怎樣？
Caused by: org.apache.kafka.common.config.ConfigException: Must set max.in.flight.requests.per.connection to at most 5 to use the idempotent producer.
推薦閱讀：
誓死要將Notepad++拉下馬，大佬推出了一款國(guó)產(chǎn)開(kāi)源編輯器.....
Linux服務(wù)器存在大量log日志，如何快速定位錯(cuò)誤？
互聯(lián)網(wǎng)初中高級(jí)大廠面試題(9個(gè)G) 內(nèi)容包含Java基礎(chǔ)、JavaWeb、MySQL性能優(yōu)化、JVM、鎖、百萬(wàn)并發(fā)、消息隊(duì)列、高性能緩存、反射、Spring全家桶原理、微服務(wù)、Zookeeper......等技術(shù)棧！
?戳閱讀原文領(lǐng)取！ 朕已閱
