小编点评

以下是Kafka事务中使用到的5个API的归纳总结： **1. 初始化事务** * **功能**：初始化事务，为后续操作做好准备。 * **API**：`void initTransactions()` **2. 开启事务** * **功能**：正式开始一个新的事务。 * **API**：`void beginTransaction() throws ProducerFencedException` **3. 在事务内提交已经消费的偏移量** * **功能**：将消费者已消费的偏移量提交到事务中。 * **API**：`void sendOffsetsToTransaction(Map offsets, String consumerGroupId) throws ProducerFencedException` **4. 提交事务** * **功能**：提交事务，确认事务中的所有操作都已成功执行。 * **API**：`void commitTransaction() throws ProducerFencedException` **5. 放弃事务** * **功能**：放弃事务，撤销事务中的所有操作。 * **API**：`void abortTransaction() throws ProducerFencedException` 以上是Kafka事务中使用到的5个API的简要总结。这些API共同确保了事务的原子性、一致性和持久性，从而保证了数据的一致性和可靠性。

正文

流程

kafka事务使用的5个API

// 1. 初始化事务
void initTransactions();
// 2. 开启事务
void beginTransaction() throws ProducerFencedException;
// 3. 在事务内提交已经消费的偏移量(主要用于消费者)
void sendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException;
// 4. 提交事务
void commitTransaction() throws ProducerFencedException;
// 5. 放弃事务(类似回滚事务的操作)
void abortTransaction() throws ProducerFencedException;

（1）初始化事务
生产者带着自定义的transactional.id发起initTransactions()，事务协调器会分配一个Pid给生产者，同幂等性一样；

（2）开启事务
生产者调用beginTransaction()开启事务，事务协调器会记录事务状态并保存在事务日志中，事务日志是存储与内部topic __transaction_state里的，对消费者不可见；

（3）提交位移Offset
生产者发送OffsetCommitRequest 请求到消费组协调者，消费组协调者会把 offset 存储到 Kafka 内部主题 __consumer_offsets 中。

（4）提交事务or回滚事务
生产者在完成消息发送后，决定提交事务
调用commitTransaction()或者abortTransaction()
事务协调器会向每个分区Leader写入一个控制消息（abort或者commit），会先发送EndTransaction请求给事务协调器
TC会向内部主题写入prepare_commit或者prepare_abort消息，后会发送给主题分区Leader一个控制消息commit或者abort，告知消息是接受还是丢弃；
如果事务成功提交，事务的消息将对消费者可见，而中止的消息会不可见，消费者在read_committed隔离级别下对于中止消息是不可见的。