Apache Kafka を使用して Java 分散トランザクションを実装する方法

Apache Kafka は Java 分散トランザクションをサポートします: トランザクションを有効にする: プロデューサとコンシューマのトランザクション プロパティを構成します。トランザクションの処理: トランザクション インターフェイスを使用して、メッセージを送信し、トランザクションをコミットまたはロールバックします。実際のケース: Kafka トランザクションを使用して注文情報をアトミックに送信し、異なるシステム間のデータの一貫性を確保します。注: トランザクションはパーティションごとに分離されており、パフォーマンスが低下する可能性があります。キーはトランザクションを識別して競合を回避するために使用されます。

Apache Kafka を使用して Java 分散トランザクションを実装する方法

はじめに

Apache Kafka は、高スループット、低遅延の分散メッセージ送信ソリューションを提供するストリーム処理プラットフォームです。トランザクション サポートが組み込まれているため、分散環境でのデータの一貫性を確保できます。この記事では、Apache Kafka と Java API を使用して分散トランザクションを実装する方法について説明します。

依存関係

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

Kafka トランザクションを設定する

Kafka トランザクションを使用するには、プロデューサー トランザクションとコンシューマ トランザクションを有効にする必要があります:

Properties properties = new Properties();
properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transaction-id");

// 创建生产者
Producer<String, String> producer = new KafkaProducer<>(properties);

// 开始事务
producer.initTransactions();

Properties properties = new Properties();
properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

// 创建消费者
Consumer<String, String> consumer = new KafkaConsumer<>(properties);

// 订阅主题
consumer.subscribe(Arrays.asList("my-topic"));

トランザクション レコードを処理する

トランザクションでは、transactional を使用する必要があります。インターフェースはメッセージを送信し、トランザクションをコミットまたはロールバックします:

// 发消息
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("my-topic", "key1", "value1"));
    producer.send(new ProducerRecord<>("my-topic", "key2", "value2"));

    // 提交事务
    producer.commitTransaction();

} catch (Exception e) {
    producer.abortTransaction();
}

// 拉取消息
try {
    consumer.subscribe(Arrays.asList("my-topic"));
    ConsumerRecords<String, String> records = consumer.poll(100);

    for (ConsumerRecord<String, String> record : records) {
        // 处理消息
    }

    // 提交偏移量，避免重复消费
    consumer.commitSync();

} catch (Exception e) {
    consumer.seekToBeginning(consumer.assignment());
}

実際的なケース

注文情報をあるシステムから別のシステムに転送する必要があるアプリケーションがあるとします。注文情報がアトミックに送信されるようにするには、Apache Kafka と分散トランザクションを使用して次のことを実現します。

1. 注文システムで、Kafka プロデューサ トランザクションを使用して注文情報を送信します。
2. 受信システムでは、Kafka コンシューマー トランザクションを使用して注文情報をプルし、処理します。
3. 注文が正常に処理された場合は、消費者トランザクションを送信して、注文情報が受信システムのデータベースに保持されていることを確認します。
4. 注文処理が失敗した場合は、コンシューマトランザクションをロールバックし、注文情報のプルをキャンセルします。

こうすることで、システム障害やネットワークの問題が発生した場合でも、2 つのシステム間で注文情報の一貫性を確保できます。

Notes

• Apache Kafka のトランザクションはパーティションごとに分離されます。つまり、単一のパーティションへのコミットは他のパーティションに影響を与えません。
• トランザクションを使用する場合、Kafka はトランザクションのメタデータを維持する必要があるため、パフォーマンスが低下する可能性があります。
• 複数のトランザクションが競合しないように、トランザクションを一意に識別するために使用される部分に Kafka レコード キーを設定してください。

以上がApache Kafka を使用して Java 分散トランザクションを実装する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。