前言

現(xiàn)在假定這么一個(gè)業(yè)務(wù)場景，從kafka中的topic獲取消息數(shù)據(jù)，經(jīng)過一定加工處理后，發(fā)送到另外一個(gè)topic中，要求整個(gè)過程消息不能丟失，也不能重復(fù)發(fā)送，即實(shí)現(xiàn)端到端的Exactly-Once精確一次消息投遞。這該如何實(shí)現(xiàn)呢？

Kafka事務(wù)介紹

針對(duì)上面的業(yè)務(wù)場景，kafka已經(jīng)替我們想到了，在kafka 0.11版本以后，引入了一個(gè)重大的特性：冪等性和事務(wù)。

冪等性

這里提到冪等性的原因，主要是因?yàn)槭聞?wù)的啟用必須要先開啟冪等性，那么什么是冪等性呢？

冪等性是指生產(chǎn)者無論向kafka broker發(fā)送多少次重復(fù)的數(shù)據(jù)，broker 端只會(huì)持久化一條，保證數(shù)據(jù)不會(huì)重復(fù)。

冪等性通過生產(chǎn)者配置項(xiàng)enable.idempotence=true開啟，默認(rèn)情況下為true。

冪等性實(shí)現(xiàn)原理

1. 每條消息都有一個(gè)主鍵，這個(gè)主鍵由 <PID, Partition, SeqNumber>組成。

• PID：ProducerID，每個(gè)生產(chǎn)者啟動(dòng)時(shí)，Kafka 都會(huì)給它分配一個(gè) ID，ProducerID 是生產(chǎn)者的唯一標(biāo)識(shí)，需要注意的是，Kafka 重啟也會(huì)重新分配 PID。
• Partition：消息需要發(fā)往的分區(qū)號(hào)。
• SeqNumber：生產(chǎn)者，他會(huì)記錄自己所發(fā)送的消息，給他們分配一個(gè)自增的 ID，這個(gè) ID 就是 SeqNumber，是該消息的唯一標(biāo)識(shí)，每發(fā)送一條消息，序列號(hào)加 1。

2. 對(duì)于主鍵相同的數(shù)據(jù)，kafka 是不會(huì)重復(fù)持久化的，它只會(huì)接收一條。

冪等性缺點(diǎn)

根據(jù)冪等性的原理，我們發(fā)現(xiàn)它存在下面的缺點(diǎn):

• 只能保證單分區(qū)、單會(huì)話內(nèi)的數(shù)據(jù)不重復(fù)
• kafka 掛掉，重新給生產(chǎn)者分配了 PID，還是有可能產(chǎn)生重復(fù)的數(shù)據(jù)

那么如何實(shí)現(xiàn)跨分區(qū)、kafka broker重啟也能保證不重復(fù)呢？這就要使用事務(wù)了。

事務(wù)

所謂事務(wù)，就是要求保證原子性，要么全部成功，要么全部失敗。那么具體該如何開啟呢？

1. kafka要想開啟事務(wù)必須要啟用冪等性，即生產(chǎn)者配置enable.idempotence=true
2. kafka生產(chǎn)者需要配置唯一的事務(wù)idtransactional.id, 最好為其設(shè)置一個(gè)有意義的名字。
3. kafka消費(fèi)端也有一個(gè)配置項(xiàng)isolation.level和事務(wù)有很大關(guān)系。

• read_uncommitted：默認(rèn)值，消費(fèi)端應(yīng)用可以看到（消費(fèi)到）未提交的事務(wù)，當(dāng)然對(duì)于已提交的事務(wù)也是可見的。
• read_committed：消費(fèi)端應(yīng)用只能消費(fèi)到提交的事務(wù)內(nèi)的消息。

Kafka事務(wù) API

現(xiàn)在我們用java的api來實(shí)現(xiàn)一下前面這個(gè)“消費(fèi)-處理-生產(chǎn)“的例子吧。

1. 引入依賴

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.4.0</version>
</dependency>

2. 創(chuàng)建事務(wù)的生產(chǎn)者

Properties prodcuerProps = new Properties();
// kafka地址
prodcuerProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
// key序列化
prodcuerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// value序列化
prodcuerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
// 啟用冪等性
producerProps.put("enable.idempotence", "true");
// 設(shè)置事務(wù)id
producerProps.put("transactional.id", "prod-1");
KafkaProducer<String, String> producer = new KafkaProducer(prodcuerProps);

• enable.idempotence配置項(xiàng)目為true
• 設(shè)置transactional.id

3. 創(chuàng)建事務(wù)的消費(fèi)者

Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); 
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put("group.id", "my-group-id");
// 設(shè)置consumer手動(dòng)提交
consumerProps.put("enable.auto.commit", "false");
// 設(shè)置隔離級(jí)別，讀取事務(wù)已提交的消息
consumerProps.put("isolation.level", "read_committed");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
//訂閱主題
consumer.subscribe(Collections.singletonList("topic1"));

• enable.auto.commit=false,設(shè)置手動(dòng)提交消費(fèi)者offset
• 設(shè)置isolation.level=read_committed，消費(fèi)事務(wù)已提交的消息

4.核心邏輯

// 初始化事務(wù) 
producer.initTransactions();
while(true) {
 // 拉取消息 
 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
    if(!records.isEmpty()){
        // 準(zhǔn)備一個(gè) hashmap 來記錄："分區(qū)-消費(fèi)位移" 鍵值對(duì)
        HashMap<TopicPartition, OffsetAndMetadata> offsetsMap = new HashMap<>();
        // 開啟事務(wù) 
        producer.beginTransaction();
        try {
            // 獲取本批消息中所有的分區(qū)
            Set<TopicPartition> partitions = records.partitions();
            // 遍歷每個(gè)分區(qū)
            for (TopicPartition partition : partitions) {
                // 獲取該分區(qū)的消息
                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                // 遍歷每條消息
                for (ConsumerRecord<String, String> record : partitionRecords) {
                    // 執(zhí)行數(shù)據(jù)的業(yè)務(wù)處理邏輯
                    ProducerRecord<String, String> outRecord = new ProducerRecord<>("topic2", record.key(), record.value().toUpperCase());
                    // 將處理結(jié)果寫入 kafka
                    producer.send(outRecord);
                }

                // 將處理完的本分區(qū)對(duì)應(yīng)的消費(fèi)位移記錄到 hashmap 中
                long offset = partitionRecords.get(partitionRecords.size() - 1).offset();
                // 事務(wù)提交的是即將到來的偏移量，這意味著我們需要加 1
                offsetsMap.put(partition,new OffsetAndMetadata(offset+1));
            }
            // 向事務(wù)管理器提交消費(fèi)位移 
            producer.sendOffsetsToTransaction(offsetsMap,"groupid");
            // 提交事務(wù) 
            producer.commitTransaction();
        } catch(Exeception e) {
            e.printStackTrace();
            // 終止事務(wù) 
            producer.abortTransaction();
        }
    }
}

• initTransactions(): 初始化事務(wù)
• beginTransaction(): 開啟事務(wù)
• sendOffsetsToTransaction(): 在事務(wù)內(nèi)提交已經(jīng)消費(fèi)的偏移量（主要用于消費(fèi)者）
• commitTransaction(): 提交事務(wù)
• abortTransaction(): 放棄事務(wù)

Kafka事務(wù)實(shí)現(xiàn)原理

kafka事務(wù)的實(shí)現(xiàn)引入了事務(wù)協(xié)調(diào)器，如下圖所示：

1. 生產(chǎn)者使用事務(wù)必須配置事務(wù)id, kafka根據(jù)事務(wù)id計(jì)算分配事務(wù)協(xié)調(diào)器
2. 事務(wù)協(xié)調(diào)器返回pid，前面的冪等性中需要
3. 開始發(fā)送消息到topic中，不過這些消息與普通的消息不同，它們帶著一個(gè)字段標(biāo)識(shí)自己是事務(wù)消息
4. 當(dāng)生產(chǎn)者事務(wù)內(nèi)的消息發(fā)送完畢，會(huì)向事務(wù)協(xié)調(diào)器發(fā)送 commit 或 abort 請(qǐng)求，等待 kafka 響應(yīng)
5. 事務(wù)協(xié)調(diào)器收到請(qǐng)求后先持久化到內(nèi)置事務(wù)主題__transaction_state中，__transaction_state默認(rèn)有50個(gè)分區(qū)，每個(gè)分區(qū)負(fù)責(zé)一部分事務(wù)。事務(wù)劃分是根據(jù)transactional.id的hashcode值%50，計(jì)算出該事務(wù)屬于哪個(gè)分區(qū)。 該分區(qū)Leader副本所在的broker節(jié)點(diǎn)即為這個(gè)transactional.id對(duì)應(yīng)的Transaction Coordinator節(jié)點(diǎn)，這也是上面第一步中的計(jì)算邏輯。
6. 事務(wù)協(xié)調(diào)器后臺(tái)會(huì)跟topic通信，告訴它們事務(wù)是成功還是失敗的。

• 如果是成功，topic會(huì)匯報(bào)自己已經(jīng)收到消息，協(xié)調(diào)者收到主題的回應(yīng)便確認(rèn)了事務(wù)完成，并持久化這一結(jié)果。
• 如果是失敗的，主題會(huì)把這個(gè)事務(wù)內(nèi)的消息丟棄，并匯報(bào)給協(xié)調(diào)者，協(xié)調(diào)者收到所有結(jié)果后再持久化這一信息，事務(wù)結(jié)束。

7. 持久化第6步中的事務(wù)成功或者失敗的信息, 如果kafka broker配置max.transaction.timeout.ms之前既不提交也不中止事務(wù), kafka broker將中止事務(wù)本身。 此屬性的默認(rèn)值為 15 分鐘。

總結(jié)

本文講解了通過kafka事務(wù)可以實(shí)現(xiàn)端到端的精確一次的消息語義，通過事務(wù)機(jī)制，KAFKA 實(shí)現(xiàn)了對(duì)多個(gè) topic 的多個(gè) partition 的原子性的寫入，通過一個(gè)例子了解了一下如何使用事物。同時(shí)也簡單介紹了事務(wù)實(shí)現(xiàn)的原理，它底層必須要依賴kafka的冪等性機(jī)制，同時(shí)通過類似“二段提交”的方式保證事務(wù)的原子性。