大家好，我是哪吒。

前兩天，有個(gè)朋友去面試，被問到Kafka事務(wù)的問題。

她的第一反應(yīng)是：

我是來面試Java的，怎么問我大數(shù)據(jù)的Kafka？

文末送5本《Spring Boot 3核心技術(shù)與最佳實(shí)踐》

不過Kafka確實(shí)是Java程序員必備的中間件技術(shù)了，這點(diǎn)是毋庸置疑的。

Kafka幾乎是當(dāng)今時(shí)代背景下數(shù)據(jù)管道的首選，無論你是做后端開發(fā)、還是大數(shù)據(jù)開發(fā)，對它可能都不陌生。開源軟件Kafka的應(yīng)用越來越廣泛。

面對Kafka的普及和學(xué)習(xí)熱潮，哪吒想分享一下自己多年的開發(fā)經(jīng)驗(yàn)，帶領(lǐng)讀者比較輕松地掌握Kafka的相關(guān)知識(shí)

上一節(jié)我們說到了解密Kafka主題的分區(qū)策略：提升實(shí)時(shí)數(shù)據(jù)處理的關(guān)鍵，今天系統(tǒng)的說一下Kafka的事務(wù)，實(shí)現(xiàn)步步為營，逐個(gè)擊破，拿下Kafka。

在當(dāng)今大數(shù)據(jù)時(shí)代，數(shù)據(jù)的可靠性和一致性變得至關(guān)重要。Kafka作為一個(gè)分布式流數(shù)據(jù)平臺(tái)，強(qiáng)調(diào)了實(shí)時(shí)數(shù)據(jù)的高吞吐量傳輸，而Kafka事務(wù)性消息則在這個(gè)過程中發(fā)揮了至關(guān)重要的作用。

本文將詳細(xì)介紹Kafka事務(wù)性消息，探究它們?nèi)绾未_保數(shù)據(jù)一致性，以及在各種應(yīng)用場景中的應(yīng)用。

一、Kafka事務(wù)性消息

1.1 介紹Kafka事務(wù)性消息

Kafka事務(wù)性消息是一項(xiàng)關(guān)鍵的功能，為確保數(shù)據(jù)一致性提供了重要的支持。在本部分，我們將深入了解Kafka事務(wù)性消息的基本概念。

Kafka事務(wù)性消息的概念

Kafka事務(wù)性消息是一種機(jī)制，用于確保消息的可靠性傳遞和處理。與非事務(wù)性消息相比，它們在數(shù)據(jù)處理中提供了額外的保證。一旦消息被寫入Kafka集群，它們將被認(rèn)為是已經(jīng)處理，無論發(fā)生了什么。

為什么需要事務(wù)性消息？

事務(wù)性消息對于確保數(shù)據(jù)一致性至關(guān)重要。在某些應(yīng)用程序中，消息的完整性和可靠性至關(guān)重要。如果在消息處理期間發(fā)生故障，如何保證消息不會(huì)丟失或重復(fù)是一個(gè)復(fù)雜的問題。Kafka事務(wù)性消息提供了解決這些問題的方式，使得消息處理更加可控和可靠。

事務(wù)性消息的特性

Kafka事務(wù)性消息具有以下關(guān)鍵特性：

• 原子性：事務(wù)性消息要么完全成功，要么完全失敗。這確保了消息不會(huì)被部分處理。

• 可靠性：一旦消息被寫入Kafka，它們將被視為已經(jīng)處理，即使發(fā)生了應(yīng)用程序或系統(tǒng)故障。

• 順序性：事務(wù)性消息在單個(gè)分區(qū)內(nèi)保持順序。這對于需要按順序處理的應(yīng)用程序至關(guān)重要。

• 冪等性：Kafka生產(chǎn)者可以配置為冪等，確保相同的消息不會(huì)被重復(fù)發(fā)送。

• Exactly Once語義：事務(wù)性消息支持"僅一次"語義，即消息要么完全到達(dá)一次，要么不到達(dá)。

本節(jié)的目標(biāo)是幫助您理解Kafka事務(wù)性消息的核心概念。接下來，我們將探討它們的應(yīng)用場景以及相對于非事務(wù)性消息的優(yōu)勢。

1.2 事務(wù)性消息的應(yīng)用場景

事務(wù)性消息在多種應(yīng)用場景中發(fā)揮著關(guān)鍵作用。以下是一些常見的應(yīng)用場景，其中事務(wù)性消息特別有用：

金融交易處理：在金融領(lǐng)域，每筆交易都必須具備原子性，確保不發(fā)生不一致或重復(fù)的交易。事務(wù)性消息可用于記錄和處理金融交易，保證交易的完整性。

訂單處理：在電子商務(wù)平臺(tái)上，訂單處理必須是可靠的，以確保訂單的創(chuàng)建、支付和發(fā)貨不會(huì)出現(xiàn)問題。事務(wù)性消息可用于跟蹤和處理訂單的不同階段，從而確保訂單流程的一致性。

庫存管理：對于企業(yè)，庫存管理是至關(guān)重要的。事務(wù)性消息可用于跟蹤庫存的變化，以確保庫存的準(zhǔn)確性和可靠性。

日志記錄：在大數(shù)據(jù)和日志記錄應(yīng)用中，日志的完整性是至關(guān)重要的。事務(wù)性消息可用于確保日志的完整性，即使在日志處理集群發(fā)生故障時(shí)也能保持一致性。

系統(tǒng)通知：對于需要向用戶發(fā)送通知或提醒的應(yīng)用程序，確保通知的可靠發(fā)送至關(guān)重要。事務(wù)性消息可用于實(shí)現(xiàn)這一目標(biāo)。

1.3 Kafka事務(wù)性消息的優(yōu)勢

相對于非事務(wù)性消息，Kafka事務(wù)性消息具有明顯的優(yōu)勢，特別是在需要數(shù)據(jù)一致性的應(yīng)用場景中。以下是Kafka事務(wù)性消息的優(yōu)勢：

數(shù)據(jù)一致性：事務(wù)性消息可確保消息要么被完全處理，要么不被處理。這消除了數(shù)據(jù)處理中的不一致性，有助于維護(hù)數(shù)據(jù)一致性。

可靠性：一旦消息被寫入Kafka，它們將被視為已經(jīng)處理，即使發(fā)生了應(yīng)用程序或系統(tǒng)故障。這確保了消息的可靠傳遞。

冪等性：Kafka生產(chǎn)者可以配置為冪等，這意味著相同的消息不會(huì)被重復(fù)發(fā)送。這有助于減少不必要的消息傳遞，避免數(shù)據(jù)重復(fù)。

Exactly Once語義：事務(wù)性消息支持"僅一次"語義，即消息要么完全到達(dá)一次，要么不到達(dá)。這是某些應(yīng)用程序所需的高級(jí)語義。

錯(cuò)誤處理：事務(wù)性消息提供了一種處理錯(cuò)誤的機(jī)制，以確保消息可以被恢復(fù)或重試，而不會(huì)丟失。

二、Kafka事務(wù)性消息的使用

在這一部分，我們將深入研究如何使用Kafka事務(wù)性消息來確保數(shù)據(jù)的一致性。

2.1 配置Kafka以支持事務(wù)性消息

配置Kafka以支持事務(wù)性消息對于確保消息在傳遞和處理過程中的一致性非常重要。在本節(jié)中，我們將詳細(xì)討論如何配置Kafka以支持事務(wù)性消息，包括生產(chǎn)者和消費(fèi)者的設(shè)置。

生產(chǎn)者配置

在生產(chǎn)者端，需要進(jìn)行一些特定的配置以啟用事務(wù)性消息。以下是一些關(guān)鍵的配置參數(shù)：

• acks：這是有關(guān)生產(chǎn)者接收到確認(rèn)之后才認(rèn)為消息發(fā)送成功的設(shè)置。對于事務(wù)性消息，通常將其設(shè)置為acks=all，以確保消息僅在事務(wù)完全提交后才被視為成功發(fā)送。

• transactional.id：這是用于標(biāo)識(shí)生產(chǎn)者實(shí)例的唯一ID。在配置文件中設(shè)置transactional.id是啟用事務(wù)性消息的關(guān)鍵步驟。

• enable.idempotence：冪等性是指相同的消息不會(huì)被重復(fù)發(fā)送。對于事務(wù)性消息，通常將其設(shè)置為enable.idempotence=true，以確保消息不會(huì)重復(fù)發(fā)送。

配置示例：

acks=all
transactional.id=my-transactional-id
enable.idempotence=true

消費(fèi)者配置

在消費(fèi)者端，同樣需要進(jìn)行適當(dāng)?shù)呐渲靡源_保正確處理事務(wù)性消息。以下是一些消費(fèi)者的重要配置參數(shù)：

• isolation.level：這是用于控制消費(fèi)者的隔離級(jí)別的設(shè)置。對于事務(wù)性消息，通常將其設(shè)置為isolation.level=read_committed，以確保只讀取已經(jīng)提交的事務(wù)消息。

• auto.offset.reset：這是消費(fèi)者啟動(dòng)時(shí)從哪里開始讀取消息的設(shè)置。通常將其設(shè)置為auto.offset.reset=earliest，以確保不會(huì)錯(cuò)過任何已提交的消息。

配置示例：

isolation.level=read_committed
auto.offset.reset=earliest

配置Kafka以支持事務(wù)性消息是確保消息可靠傳遞和處理的關(guān)鍵步驟。這些配置設(shè)置可以確保在生產(chǎn)和消費(fèi)事務(wù)性消息時(shí)的正確行為。

2.2 生產(chǎn)者：發(fā)送事務(wù)性消息

在這一部分，我們將深入研究如何使用Kafka生產(chǎn)者來發(fā)送事務(wù)性消息。發(fā)送事務(wù)性消息是確保數(shù)據(jù)一致性的關(guān)鍵步驟，需要特別小心。以下是詳細(xì)的步驟和示例：

創(chuàng)建Kafka生產(chǎn)者

首先，我們需要?jiǎng)?chuàng)建一個(gè) Kafka 生產(chǎn)者的實(shí)例。這個(gè)生產(chǎn)者實(shí)例將負(fù)責(zé)將消息發(fā)送到 Kafka 主題。創(chuàng)建生產(chǎn)者需要配置參數(shù)，包括 Kafka 集群的地址、消息的鍵和值的序列化器、事務(wù)ID 等。

下面是一個(gè)創(chuàng)建 Kafka 生產(chǎn)者的示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

public class MyKafkaProducer {
    public static Producer<String, String> createProducer() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");

        return new KafkaProducer<>(properties);
    }
}

開始事務(wù)

在準(zhǔn)備發(fā)送事務(wù)性消息之前，我們需要明確地開始一個(gè)事務(wù)。這通過調(diào)用 beginTransaction 方法來實(shí)現(xiàn)。一旦事務(wù)開始，所有后續(xù)的消息發(fā)送將包含在這個(gè)事務(wù)中。

producer.beginTransaction();

發(fā)送消息

在事務(wù)內(nèi)，我們可以開始發(fā)送消息。這些消息將被包含在事務(wù)中，只有在事務(wù)成功提交時(shí)才會(huì)真正寫入 Kafka 主題。

producer.send(new ProducerRecord<>("my-topic", "key1", "value1"));
producer.send(new ProducerRecord<>("my-topic", "key2", "value2"));

提交或中止事務(wù)

事務(wù)性消息的一個(gè)關(guān)鍵特性是它們要么完全成功，要么完全失敗。因此，在消息發(fā)送后，我們需要根據(jù)消息的處理結(jié)果來決定是提交事務(wù)還是中止事務(wù)。這可以通過調(diào)用 commitTransaction 或 abortTransaction 方法來實(shí)現(xiàn)。

try {
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // 處理異常，通常中止事務(wù)并重試
    producer.close();
} catch (CommitFailedException e) {
    // 事務(wù)提交失敗，通常中止事務(wù)并重試
    producer.close();
}

上述步驟提供了一個(gè)基本的示例，演示如何使用 Kafka 生產(chǎn)者發(fā)送事務(wù)性消息。事務(wù)性消息的發(fā)送確保了消息的可靠性和一致性，尤其在需要原子性保證的情況下非常有用。

2.3 消費(fèi)者：處理事務(wù)性消息

在這一部分，我們將深入研究如何使用 Kafka 消費(fèi)者來處理事務(wù)性消息。正確處理事務(wù)性消息對于保證數(shù)據(jù)一致性至關(guān)重要。以下是詳細(xì)的步驟和示例：

創(chuàng)建 Kafka 消費(fèi)者

首先，我們需要?jiǎng)?chuàng)建一個(gè) Kafka 消費(fèi)者的實(shí)例。這個(gè)消費(fèi)者實(shí)例將負(fù)責(zé)從 Kafka 主題中讀取消息。創(chuàng)建消費(fèi)者需要配置參數(shù)，包括 Kafka 集群的地址、消息的鍵和值的反序列化器、消費(fèi)者組 ID 等。

下面是一個(gè)創(chuàng)建 Kafka 消費(fèi)者的示例：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class MyKafkaConsumer {
    public static Consumer<String, String> createConsumer() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        return new KafkaConsumer<>(properties);
    }
}

訂閱主題

消費(fèi)者需要明確地訂閱包含事務(wù)性消息的主題。這通過調(diào)用 subscribe 方法來實(shí)現(xiàn)。一旦訂閱，消費(fèi)者將開始接收該主題上的消息。

consumer.subscribe(Collections.singletonList("my-topic"));

處理消息

一旦事務(wù)性消息到達(dá)，消費(fèi)者需要確保消息被正確處理。這通常涉及到處理消息的邏輯，確保數(shù)據(jù)的一致性。處理消息的邏輯將根據(jù)具體的應(yīng)用和需求而異。

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    String key = record.key();
    String value = record.value();
    // 處理消息的邏輯
}

提交位移

消費(fèi)者需要負(fù)責(zé)提交消息的位移，以便正確跟蹤已處理的消息。這通過調(diào)用 commitSync 或 commitAsync 方法來實(shí)現(xiàn)。位移的提交確保了消息不會(huì)被重復(fù)處理。

consumer.commitSync();

上述步驟提供了一個(gè)基本的示例，演示了如何使用 Kafka 消費(fèi)者處理事務(wù)性消息。消費(fèi)者的正確配置和消息處理確保了消息的可靠性和一致性。在實(shí)際應(yīng)用中，處理消息的邏輯將更加復(fù)雜，以滿足特定的需求。

三、事務(wù)性消息的最佳實(shí)踐

在這一節(jié)，我們將提供一些關(guān)于如何使用Kafka事務(wù)性消息的最佳實(shí)踐。這包括如何確保消息的一次交付、監(jiān)控和故障排查以及性能優(yōu)化。

3.1 保障消息的一次交付

3.1.1 生產(chǎn)者冪等性

確保生產(chǎn)者的冪等性是關(guān)鍵，以防止消息被重復(fù)發(fā)送。以下是一些關(guān)鍵策略和實(shí)踐，可用于確保生產(chǎn)者的冪等性：

1. 分配唯一消息ID： 為每條消息分配一個(gè)唯一的消息ID。這可以是全局唯一的，也可以是特定于主題的唯一。在發(fā)送消息之前，生產(chǎn)者可以檢查已經(jīng)發(fā)送的消息記錄，以確保當(dāng)前消息的ID不重復(fù)。

2. 使用冪等性API： Kafka 提供了冪等性的生產(chǎn)者 API。你可以在生產(chǎn)者配置中啟用冪等性，設(shè)置 enable.idempotence=true，以確保消息在發(fā)送時(shí)不會(huì)被重復(fù)處理。

3. 實(shí)現(xiàn)自定義冪等性： 在一些情況下，自定義實(shí)現(xiàn)冪等性邏輯可能是必要的。這可以涉及到在消息處理端的數(shù)據(jù)庫或存儲(chǔ)中跟蹤已處理消息的狀態(tài)，以確保消息不會(huì)被重復(fù)處理。

4. 設(shè)置適當(dāng)?shù)闹卦嚈C(jī)制： 如果消息發(fā)送失敗，生產(chǎn)者應(yīng)該具備適當(dāng)?shù)闹卦嚈C(jī)制，以確保消息最終被成功發(fā)送。重試機(jī)制需要在生產(chǎn)者的配置中進(jìn)行設(shè)置。

3.1.2 消費(fèi)者去重

保障消息不會(huì)被重復(fù)處理同樣至關(guān)重要。以下是一些策略和最佳實(shí)踐，可用于實(shí)現(xiàn)消費(fèi)者的去重：

1. 冪等性消息處理邏輯： 消費(fèi)者的消息處理邏輯應(yīng)該是冪等的。這意味著無論消息被處理多少次，其結(jié)果都應(yīng)該是相同的。這通常需要在應(yīng)用程序代碼中進(jìn)行實(shí)施。

2. 消息唯一標(biāo)識(shí)： 為每條消息分配一個(gè)唯一的標(biāo)識(shí)符，如消息ID。在處理消息前，消費(fèi)者可以維護(hù)一個(gè)記錄已處理消息的數(shù)據(jù)結(jié)構(gòu)，以確保消息不會(huì)被重復(fù)處理。

3. 消費(fèi)者去重過程： 消費(fèi)者在處理消息前，可以查詢已處理消息的記錄，如果消息已存在于記錄中，可以選擇跳過處理或進(jìn)行進(jìn)一步處理。這可以防止消息的重復(fù)處理。

4. 消費(fèi)者庫支持： 一些消息隊(duì)列處理庫提供了內(nèi)置的去重機(jī)制，你可以利用這些庫來簡化去重處理。

以上內(nèi)容提供了詳細(xì)的策略和最佳實(shí)踐，以確保消息的一次交付。這是保障數(shù)據(jù)一致性的關(guān)鍵步驟，特別適用于事務(wù)性消息的處理。這些實(shí)踐可以根據(jù)具體的應(yīng)用和需求進(jìn)行定制化。

3.2 事務(wù)性消息的監(jiān)控和故障排查

3.2.1 監(jiān)控工具

監(jiān)控Kafka事務(wù)性消息是確保系統(tǒng)的可靠性的重要部分。以下是一些監(jiān)控工具和策略：

• Kafka內(nèi)置指標(biāo)：Kafka提供了一組內(nèi)置指標(biāo)，用于監(jiān)控事務(wù)性消息的性能和狀態(tài)。你可以使用這些指標(biāo)來跟蹤消息的處理情況。

• 日志文件：Kafka的日志文件包含了詳細(xì)的事件信息，可以用于故障排查和性能分析。定期檢查日志文件，以查找潛在的問題。

• 監(jiān)控系統(tǒng)：使用專業(yè)的監(jiān)控系統(tǒng)，如Prometheus和Grafana，來建立實(shí)時(shí)監(jiān)控和警報(bào)。這些系統(tǒng)可以幫助你及時(shí)發(fā)現(xiàn)問題并采取措施。

3.2.2 故障排查

當(dāng)事務(wù)性消息出現(xiàn)問題時(shí)，需要能夠排查和解決這些問題。以下是一些故障排查策略：

• 日志分析：定期分析Kafka的日志文件，查找異常和錯(cuò)誤信息。這可以幫助你及早發(fā)現(xiàn)問題并采取措施。

• 監(jiān)控警報(bào)：建立監(jiān)控警報(bào)，以便在出現(xiàn)問題時(shí)立即收到通知。這有助于快速響應(yīng)問題。

• 版本和配置管理：確保Kafka和應(yīng)用程序的版本和配置得到正確管理。不同版本或配置的不一致可能導(dǎo)致問題。

3.3 事務(wù)性消息的性能考量

性能是任何消息系統(tǒng)的關(guān)鍵指標(biāo)，特別是對于高吞吐量和低延遲的需求。以下是一些性能考量和優(yōu)化策略：

3.3.1 性能調(diào)整

• 生產(chǎn)者性能調(diào)整：通過調(diào)整生產(chǎn)者的配置參數(shù)，如batch.size、acks等，可以優(yōu)化消息發(fā)送性能。

• 消費(fèi)者性能調(diào)整：消費(fèi)者的性能也可以通過配置參數(shù)，如max.poll.records、fetch.min.bytes等進(jìn)行調(diào)整。

3.3.2 吞吐量優(yōu)化

• 分區(qū)和并

行度**：合理地選擇分區(qū)數(shù)量和消費(fèi)者的并行度，以確保系統(tǒng)能夠處理大量事務(wù)性消息。

• 水平擴(kuò)展：如果系統(tǒng)負(fù)載增加，考慮進(jìn)行水平擴(kuò)展，增加Kafka代理和消費(fèi)者實(shí)例，以提高吞吐量。

• 網(wǎng)絡(luò)和存儲(chǔ)優(yōu)化：確保網(wǎng)絡(luò)和存儲(chǔ)基礎(chǔ)設(shè)施足夠快，以支持高吞吐量的消息傳遞。

上述最佳實(shí)踐策略和性能優(yōu)化建議可以幫助你更好地使用Kafka事務(wù)性消息，確保消息的可靠傳遞和一致性處理，同時(shí)滿足性能需求。通過仔細(xì)的配置、監(jiān)控和故障排查，你可以建立一個(gè)可靠和高性能的消息處理系統(tǒng)。

四、示例：生產(chǎn)和消費(fèi)Kafka事務(wù)性消息

在這一節(jié)，我們將提供兩個(gè)示例，詳細(xì)展示如何生產(chǎn)和消費(fèi)Kafka事務(wù)性消息。

4.1 示例1：生產(chǎn)事務(wù)性消息

示例1代碼：生產(chǎn)者

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class TransactionalProducerExample {
    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String topic = "my-transactional-topic";

        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("acks", "all");
        properties.put("enable.idempotence", "true");
        properties.put("transactional.id", "my-transactional-id");

        Producer<String, String> producer = new KafkaProducer<>(properties);

        producer.initTransactions();

        try {
            producer.beginTransaction();
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", "value");
            producer.send(record);
            producer.commitTransaction();
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            // Fenced, sequence issue, or authorization exception
            producer.close();
        } catch (KafkaException e) {
            // Handle other exceptions
            producer.close();
        }

        producer.close();
    }
}

代碼說明:

• 這個(gè)示例演示了如何創(chuàng)建一個(gè)Kafka生產(chǎn)者，配置它以支持事務(wù)性消息，并生產(chǎn)一條事務(wù)性消息。
• transactional.id是一個(gè)用于標(biāo)識(shí)生產(chǎn)者事務(wù)的唯一ID。它確保了事務(wù)性消息的一致性。
• 在try塊中，我們使用producer.beginTransaction()來啟動(dòng)一個(gè)事務(wù)，然后發(fā)送一條消息，最后使用producer.commitTransaction()來提交事務(wù)。
• 如果在事務(wù)期間發(fā)生異常，我們在catch塊中處理異常并關(guān)閉生產(chǎn)者。

4.2 示例2：消費(fèi)事務(wù)性消息

示例2代碼：消費(fèi)者

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class TransactionalConsumerExample {
    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String groupId = "my-consumer-group";
        String topic = "my-transactional-topic";

        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);
        properties.put("group.id", groupId);
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
            }
        }
    }
}

代碼說明:

• 這個(gè)示例演示了如何創(chuàng)建一個(gè)Kafka消費(fèi)者，訂閱一個(gè)主題，并消費(fèi)事務(wù)性消息。
• 消費(fèi)者將持續(xù)輪詢主題以獲取新的消息。
• 每當(dāng)有新消息可用時(shí)，它將打印出消息的鍵和值。

五、總結(jié)

本文深入探討了Kafka事務(wù)性消息的關(guān)鍵概念、應(yīng)用場景、優(yōu)勢、配置、使用以及最佳實(shí)踐。在總結(jié)中，讓我們再次強(qiáng)調(diào)一些關(guān)鍵要點(diǎn)，并展望Kafka事務(wù)性消息的未來。

• Kafka事務(wù)性消息是一種機(jī)制，用于確保消息的可靠性傳遞和處理。它們提供了額外的保證，確保消息要么完全成功，要么完全失敗。
• 應(yīng)用場景：Kafka事務(wù)性消息在金融交易、庫存管理、訂單處理等需要高可靠性和數(shù)據(jù)一致性的應(yīng)用中發(fā)揮關(guān)鍵作用。
• 優(yōu)勢：事務(wù)性消息相對于非事務(wù)性消息提供了更高的數(shù)據(jù)一致性和可靠性，支持原子性、冪等性和"僅一次"語義。
• 配置：配置Kafka以支持事務(wù)性消息包括生產(chǎn)者和消費(fèi)者的設(shè)置，如transactional.id、enable.idempotence等。
• 生產(chǎn)事務(wù)性消息：使用Kafka生產(chǎn)者，需要初始化事務(wù)、發(fā)送消息，然后提交或中止事務(wù)，以確保消息的一致性。
• 消費(fèi)事務(wù)性消息：使用Kafka消費(fèi)者，需要訂閱主題并持續(xù)輪詢以獲取消息，然后確保消息被正確處理。
• 最佳實(shí)踐：最佳實(shí)踐包括保障消息的一次交付、監(jiān)控和故障排查以及性能考量，以確保系統(tǒng)的穩(wěn)定性和高性能。

六、Spring Boot 3核心技術(shù)與最佳實(shí)踐

1、內(nèi)容介紹

京東購買鏈接：Spring Boot 3核心技術(shù)與最佳實(shí)踐

本書是一本針對Java開發(fā)人員的圖書，旨在幫助Java開發(fā)人員掌握Spring Boot的基本使用，以及深入了解Spring Boot的應(yīng)用及原理。

本書內(nèi)容由淺入深、循序漸進(jìn)，第1～5章介紹Spring Boot的基礎(chǔ)知識(shí)（基礎(chǔ)入門、配置管理、Starter、自動(dòng)配置、啟動(dòng)過程與擴(kuò)展應(yīng)用、日志管理），第6～9章介紹Spring Boot的綜合應(yīng)用（Web、數(shù)據(jù)訪問、計(jì)劃任務(wù)、緩存、消息隊(duì)列），第10～12章介紹Spring Boot應(yīng)用的附加能力（調(diào)試、單元測試、打包、部署、監(jiān)控、報(bào)警），全面覆蓋了Spring Boot的核心知識(shí)要點(diǎn)。

本書涵蓋了筆者多年的研究和實(shí)踐經(jīng)驗(yàn)，從中提煉出了核心知識(shí)要點(diǎn)，從Spring Boot的基本概念和基礎(chǔ)實(shí)踐入手，再通過大量的知識(shí)點(diǎn)分析及代碼實(shí)踐，詳細(xì)介紹如何利用Spring Boot簡化開發(fā)過程，提高開發(fā)效率。

2、作者簡介

周紅亮（英文名為John），具有多年編程開發(fā)和系統(tǒng)架構(gòu)經(jīng)驗(yàn)，在大型互聯(lián)網(wǎng)公司擔(dān)任過Java高發(fā)開發(fā)工程師、開發(fā)主管、系統(tǒng)架構(gòu)師等職位。負(fù)責(zé)并參與過多個(gè)大型分布式系統(tǒng)的設(shè)計(jì)和研發(fā)、改造等，從中積累了大量的微服務(wù)系統(tǒng)架構(gòu)經(jīng)驗(yàn)。

3、參與方式

圖書數(shù)量：本次送出 5 本《Spring Boot 3核心技術(shù)與最佳實(shí)踐》 ?。。?/p>

活動(dòng)時(shí)間：截止到 2023-12-18 21:00:00

??抽獎(jiǎng)方式：

????點(diǎn)擊下方名片，回復(fù)1218，即可參與????

??哪吒會(huì)在朋友圈公布中獎(jiǎng)名單。

名單公布時(shí)間：2023-12-18 21:10:00文章來源地址http://www.zghlxwxcb.cn/news/detail-766950.html

到了這里，關(guān)于Kafka事務(wù)是怎么實(shí)現(xiàn)的？Kafka事務(wù)消息原理詳解（文末送書）的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！