Kafka Producer 幂等与事务写入实践

--- title: Kafka Producer 幂等与事务写入实践 keywords: enable.idempotence, transactional.id, acks, max.in.flight.requests.per.connection, initTransactions description: 配置 Kafka Producer 幂等与事务写入，保证端到端一致性与去重；示例展示事务开启、发送与提交流程。 tags: - Kafka - acks - enable.idempotence - initTransactions - max.in.flight.requests.per.connection - transactional.id - 数据流 - 生产者 categories: - 文章资讯 - 技术教程 --- Java Producer 属性与事务示例： ```java import org.apache.kafka.clients.producer.*; import java.util.Properties; public class TxProducer { public static void main(String[] args) throws Exception { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "orders-tx-1"); KafkaProducer producer = new KafkaProducer<>(props); producer.initTransactions(); try { producer.beginTransaction(); producer.send(new ProducerRecord<>("orders", "id-1", "{\"amount\":9.99}")); producer.send(new ProducerRecord<>("orders", "id-2", "{\"amount\":19.99}")); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); } finally { producer.close(); } } } ```