TDMQ CKafka 版事務：分佈式環境下的消息一致性保障 Detail - kafka,消息隊列,騰訊雲騰訊雲中間件 Blog

解鎖 CKafka 事務能力的神秘面紗

在當今數字化浪潮下，分佈式系統已成為支撐海量數據處理和高併發業務的中流砥柱。但在這看似堅不可摧的架構背後，數據一致性問題卻如影隨形，時刻考驗着系統的穩定性與可靠性。

CKafka 作為分佈式流處理平台的佼佼者，以其高吞吐量、可擴展性和容錯性等特點備受青睞。而它的事務功能，就是解決數據一致性問題的 “秘密武器”。通過事務能力，CKafka 能確保一組消息要麼全部成功寫入，要麼全部失敗回滾，就如同一個精密的齒輪組，每一個動作都協同一致，保證數據的完整性和準確性。無論是業務操作裏的多條消息同時發送，還是流場景裏“消費消息-處理-寫入消息”的鏈式操作，CKafka 事務能力都能大顯身手，為業務的穩健運行保駕護航。

接下來，就讓我們一起深入探索 CKafka 事務的奇妙世界，揭開它神秘的面紗。

事務相關概念大揭秘

在深入 CKafka 事務實踐之前，我們先來夯實基礎，全面瞭解事務相關的概念，為後續的實踐操作做好充分準備。

事務的基本概念

在 CKafka 的事務世界裏，原子性、一致性、隔離性和持久性是其核心特性，它們共同確保了事務操作的可靠性和數據的完整性。

原子性：事務中的所有操作要麼全部成功，要麼全部失敗。CKafka 確保在事務中發送的消息要麼被成功寫入到主題中，要麼不寫入。
一致性：確保事務執行前後，數據的狀態應該保持一致。
隔離性：事務之間的操作相互獨立，互不干擾。
持久性：一旦事務被提交，其結果就會永久性地保存下來，即使遭遇系統崩潰、機器宕機等極端故障，數據也不會丟失。

事務的工作流程

CKafka 事務的工作流程清晰有序，如同一場精心編排的交響樂，每個步驟都緊密相連，共同奏響數據一致性的樂章。

首先是啓動事務，生產者在發送消息之前，需要調用 initTransactions() 方法來初始化事務。
接着進入發送消息環節，生產者可以將多條消息發送到一個或多個主題，這些消息都會被標記為事務性消息。
最後是提交或中止事務階段：

如果所有消息都成功發送，生產者就會調用 commitTransaction() 方法來提交事務，此時所有消息將被正式寫入到 CKafka；

反之，如果在發送過程中發生錯誤，生產者可以調用 abortTransaction() 方法來中止事務，所有消息將不會被寫入。

事務的配置

要使用 CKafka 的事務功能，您需要在生產者配置中設置以下參數：

Transactional.id：是每個事務性生產者的唯一標識符，用於標識事務的所有消息，確保事務的唯一性和可追蹤性。
Acks：設置為 All，確保所有副本都確認消息。
Enable.idempotence：設置為 True ，用於啓用冪等性，確保消息不會被重複發送。

事務的限制

在使用 CKafka 事務功能過程中，您還需要注意以下限制條件：

性能開銷：使用事務會引入額外的性能開銷，因為在事務處理過程中，需要進行更多的協調和確認操作。
事務超時：CKafka 對事務有超時限制，默認情況下為 60 秒。如果事務在這個時間內未提交或中止，將會被自動中止。
消費者處理：消費者在處理事務性消息時也需要格外注意，只有在事務提交後，消費者才能看到這些消息。

事務使用示例實操

理論知識儲備完成後，接下來通過實際代碼示例，幫助您更直觀地瞭解 CKafka 事務在生產者和消費者端的具體實現方式。

Producer 示例

以下是一個使用 Java 語言編寫的 CKafka 生產者示例，展示瞭如何配置、初始化事務，發送消息並處理異常。

import org.apache.CKafka.clients.producer.CKafkaProducer;
import org.apache.CKafka.clients.producer.ProducerConfig;
import org.apache.CKafka.clients.producer.ProducerRecord;
import org.apache.CKafka.clients.producer.RecordMetadata;
import java.util.Properties;
public class TransactionalProducerDemo {
    public static void main(String[] args) {
        // CKafka 配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // CKafka broker 地址
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.CKafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.CKafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id"); // 事務 ID
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // 啓用冪等性
        // 創建 CKafka 生產者
        CKafkaProducer<String, String> producer = new CKafkaProducer<>(props);
        // 初始化事務
        producer.initTransactions();
        try {
            // 開始事務
            producer.beginTransaction();
            // 發送消息
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key-" + i, "value-" + i);
                RecordMetadata metadata = producer.send(record).get(); // 發送消息並等待確認
                System.out.printf("Sent message: key=%s, value=%s, partition=%d, offset=%d%n", 
                                  record.key(), record.value(), metadata.partition(), metadata.offset());
            }
            // 提交事務
            producer.commitTransaction();
            System.out.println("Transaction committed successfully.");
        } catch (Exception e) {
            // 如果發生異常，回滾事務
            producer.abortTransaction();
            System.err.println("Transaction aborted due to an error: " + e.getMessage());
        } finally {
            // 關閉生產者
            producer.close();
        }
    }
}

Consumer 示例

接下來是一個 CKafka 消費者示例，展示瞭如何配置並處理事務性消息，包括訂閲主題和拉取消息。

import org.apache.CKafka.clients.consumer.ConsumerConfig;
import org.apache.CKafka.clients.consumer.ConsumerRecord;
import org.apache.CKafka.clients.consumer.CKafkaConsumer;
import org.apache.CKafka.clients.consumer.ConsumerRecords;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class TransactionalConsumerDemo {
    public static void main(String[] args) {
        // CKafka 配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // CKafka broker 地址
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // 消費者組 ID
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.CKafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.CKafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); // 只讀取已提交的事務消息
        // 創建 CKafka 消費者
        CKafkaConsumer<String, String> consumer = new CKafkaConsumer<>(props);
        // 訂閲主題
        consumer.subscribe(Collections.singletonList("my-topic"));
        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed message: key=%s, value=%s, partition=%d, offset=%d%n",
                            record.key(), record.value(), record.partition(), record.offset());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 關閉消費者
            consumer.close();
        }
    }
}

CKafka 事務管理深度剖析

在 CKafka 中，事務管理涉及到多個組件和數據結構，以確保事務的原子性和一致性。事務信息的內存佔用主要與以下幾個方面有關：

事務 ID 和 Producer ID

事務 ID：每個事務都有一個唯一的事務 ID，用於標識該事務。事務 ID 是由生產者在發送消息時指定的，通常是一個字符串。
Producer ID：每個生產者在連接到 CKafka 時會被分配一個唯一的 Producer ID。這個 ID 用於標識生產者的消息，並確保消息的順序性和冪等性。

事務狀態管理

CKafka 使用一個稱為事務狀態日誌的內部主題來管理事務的狀態。這個日誌記錄了每個事務的狀態（如進行中、已提交、已中止）以及與該事務相關的消息。事務狀態日誌的管理涉及以下幾個方面：

內存中的數據結構：CKafka 在內存中維護一個數據結構（例如哈希表或映射），用於存儲當前活動的事務信息。這些信息包括事務 ID、Producer ID、事務狀態、時間戳等。
持久化存儲：事務狀態日誌會被持久化到磁盤，以確保在 CKafka 服務器重啓或故障恢復時能夠恢復事務狀態。

事務信息的內存佔用

事務信息的內存佔用主要取決於以下兩個因素：

活動事務的數量：當前正在進行的事務數量直接影響內存佔用。每個活動事務都會在內存中佔用一定的空間。
事務的元數據：每個事務的元數據（例如事務 ID、Producer ID、狀態等）也會佔用內存。具體的內存佔用量取決於這些元數據的大小。

事務的清理

為了防止內存佔用過高，CKafka 會根據配置的過期時間定期檢查並清理已完成的事務，默認保留 7 天，過期刪除。

事務常見的 FullGC / OOM 問題

從事務管理可以看出，事務信息會佔用大量內存。其中影響事務信息佔用內存大小的最直接的兩個因素就是：事務 ID 的數量和 Producer ID 的數量。

其中事務 ID 的數量指的是客户端往 Broker 初始化、提交事務的數量，這個與客户端的事務新增提交頻率強相關。
Producer ID 指的是 Broker 內每個 Topic 分區存儲的 Producer 狀態信息，因此 Producer ID 的數量與 Broker 的分區數量強相關。

在事務場景中，事務 ID 和 Producer ID 強綁定，如果同一個和事務 ID 綁定的 Producer ID 往 Broker 內所有的分區都發送消息，那麼一個 Broker 內的 Producer ID 的數量理論上最多能達到事務 ID 數量與 Broker 內分區數量的乘積。假設一個實例下的事務 ID 數量為 t，一個 Broker 下的分區數量為 p，那麼 Producer ID 的數量最大能達到 t * p。

因此，假設一個 Broker 下的事務 ID 數量為 t，平均事務內存佔用大小為 tb，一個 Broker 下的分區數量為 p，平均一個 Producer ID 佔用大小為 pb，那麼該 Broker 內存中關於事務信息佔用的內存大小為：t tb + t p * pb。

可以看出有兩種場景可能會導致內存佔用暴漲：

客户端頻繁往實例初始化新增提交新的事務 ID。
同一個事務 ID 往多個分區發送數據，Producer ID 的叉乘數量會上漲的非常恐怖，很容易將內存打滿。

因此，無論是對 Flink 客户端還是自己實現的事務 Producer，都要儘量避免這兩種場景。例如對於 Flink，可以適當降低 Checkpoint 的頻率，以減小由於事務 ID 前綴+隨機串計算的事務 ID 變化的頻率。另外就是儘量保證同一個事務 ID 往同一個分區發送數據。

Flink 使用事務注意事項

對於 Flink 有以下優化手段，來保證事務信息不會急劇膨脹：

客户端優化參數：Flink 加大 Checkpoint 間隔。
Flink 生產任務可優化 sink.partitioner 為 Fixed 模式。

Flink 參數説明：https://nightlies.apache.org/flink/flink-docs-master/zh/docs/...

總結

CKafka 事務作為分佈式系統中確保數據一致性和完整性的強大工具，為我們打開了一扇通往高效、可靠數據處理的大門。它通過原子性、一致性、隔離性和持久性的嚴格保障，以及清晰有序的工作流程，讓我們能夠在複雜的分佈式環境中，自信地處理各種數據事務，確保消息的準確傳遞和處理。

隨着分佈式系統的不斷髮展和業務需求的日益複雜，CKafka 事務必將在更多領域發揮關鍵作用。無論是金融領域的精準交易記錄，還是電商行業的訂單與庫存同步，亦或是物流系統的全程信息追蹤，CKafka 事務都將為這些業務的穩定運行提供堅實的技術支撐。

希望大家在閲讀本文後，能夠將 CKafka 事務的知識運用到實際項目中，不斷探索和實踐，在分佈式系統的開發中取得更好的成果。

騰訊雲中間件 Blog

@wangying_5e4b645338a98

Tags

serverless (33)

kafka (30)

rabbitmq (20)

客户端 (16)

消息隊列 (15)

騰訊雲 (12)

Stories

TDMQ CKafka 版事務：分佈式環境下的消息一致性保障 - Stories Detail