Kafka Producer之事务性

2025/2/22 2:25:53 来源：https://blog.csdn.net/m0_51390969/article/details/140619958 浏览: 次关键词：Kafka Producer之事务性

文章目录

1. 跨会话幂等性失效
2. 开启事务
3. 事务流程原理

事务性可以防止跨会话幂等性失效，同时也可以保证单个生产者的指定数据，要么全部成功要么全部失败，不限分区。不可以多个生产者共用相同的事务ID。

1. 跨会话幂等性失效

幂等性开启后，broker会对每个分区记录生产者状态，并且生产者具有PID，消息被标记为PID加上序列号，数据重复和有序都是在其基础之上运作的。

生产者重启等因素会导致PID变化，导致幂等性短暂失效。

2. 开启事务

因为事务是基于幂等性的，所以幂等性的配置都要有。

package org.dragon.producer;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.HashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class KafkaProducerTransactionTest {public static void main(String[] args) throws InterruptedException, ExecutionException {//创建producerHashMap<String, Object> config = new HashMap<>();config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//配置acks等级config.put(ProducerConfig.ACKS_CONFIG, "-1");config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);config.put(ProducerConfig.RETRIES_CONFIG, 5);// 把buffer改小一点，让测试数据组成更多batchconfig.put(ProducerConfig.BATCH_SIZE_CONFIG, 5);config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);// 事务IDconfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-tx-id");KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);//初始化事务producer.initTransactions();try {// 开启事务producer.beginTransaction();for (int i = 0; i < 10; i++) {//创建recordProducerRecord<String, String> record = new ProducerRecord<String, String>("test2","" + i,"我是你爹" + i);//发送recordFuture<RecordMetadata> send = producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {System.out.println("回调信息：消息发送成功");}});System.out.println("发送数据");send.get();}// 提交事务producer.commitTransaction();}catch(Exception e) {// 中止事务producer.abortTransaction();e.printStackTrace();}finally{//关闭producerproducer.close();}}
}