Comment implémenter des transactions distribuées Java à l'aide d'Apache Kafka

Apache Kafka prend en charge les transactions distribuées Java : activez les transactions : configurez les propriétés des transactions du producteur et du consommateur. Traitement des transactions : utilisez l'interface transactionnelle pour envoyer des messages et valider ou annuler des transactions. Cas pratique : Utiliser les transactions Kafka pour transmettre de manière atomique les informations de commande afin d'assurer la cohérence des données entre les différents systèmes. REMARQUE : les transactions sont isolées par partition, les performances peuvent être réduites, les clés sont utilisées pour identifier les transactions et éviter les conflits.

Comment utiliser Apache Kafka pour implémenter des transactions distribuées Java

Introduction

Apache Kafka est une plate-forme de traitement de flux qui fournit une solution de transmission de messages distribuée à haut débit et à faible latence. Il dispose d'une prise en charge intégrée des transactions, vous permettant de garantir la cohérence des données dans un environnement distribué. Cet article vous expliquera comment implémenter des transactions distribuées à l'aide d'Apache Kafka et de l'API Java.

Dépendances

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

Configurer les transactions Kafka

Pour utiliser les transactions Kafka, vous devez activer les transactions des producteurs et des consommateurs :

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"));

Traiter les enregistrements de transactions

Dans les transactions, vous devez utiliser transactional Le l'interface envoie des messages et valide ou annule des transactions :

// 发消息
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());
}

Cas pratique

Supposons que vous ayez une application qui doit transférer les informations de commande d'un système à un autre. Pour garantir que les informations de commande sont soumises de manière atomique, vous pouvez utiliser Apache Kafka et des transactions distribuées pour réaliser :

1. Dans le système de commande, utilisez les transactions du producteur Kafka pour envoyer les informations de commande.
2. Dans le système de réception, utilisez les transactions des consommateurs Kafka pour extraire les informations de commande et les traiter.
3. Si la commande est traitée avec succès, soumettez la transaction du consommateur pour vous assurer que les informations de commande sont conservées dans la base de données du système de réception.
4. Si le traitement de la commande échoue, annulez la transaction du consommateur et annulez l'extraction des informations de commande.

De cette façon, vous pouvez vous assurer que les informations de votre commande sont cohérentes entre les deux systèmes, même en cas de panne du système ou de problème de réseau.

Notes

• Les transactions dans Apache Kafka sont isolées par partition, ce qui signifie que les validations sur une seule partition n'affecteront pas les autres partitions.
• Lors de l'utilisation de transactions, les performances peuvent être réduites car Kafka doit conserver les métadonnées des transactions.
• Assurez-vous de définir la clé d'enregistrement Kafka sur la partie utilisée pour identifier de manière unique la transaction afin de garantir que plusieurs transactions n'entrent pas en conflit.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!