不知道从什么时候开始，业界刮起了一股“万物皆可 Exactly-Once”的风气。前阵子听信了某架构大厂的宣传，说我们的分布式记账系统必须无脑切成 Kafka 事务性生产，以达到端到端的精确一次性。结果赶鸭子上架折腾了两周，刚上线就发生了大面积的重复记账和脏数据。仔细排查后我们彻底无语了：原来消费端代码的 isolation.level 居然保持着默认的 read_uncommitted！这等于脱裤子放屁，事务性生产者发送的未提交/已终止事务数据，全部被消费端直接脏读了。

要实现 Kafka 端到端的一次性语义（EOS，Exactly-Once Semantics），必须满足三个核心条件，缺一不可：

  1. 幂等性生产者（Idempotent Producer）：通过 enable.idempotence = true 避免网络抖动重试引起的重复消息。
  2. 事务性生存周期（Transactional API）：通过 transactional.id 将一组发往不同 Partition 的消息打包在一个事务中。
  3. 读已提交隔离（Read Committed）：这是最容易遗漏的，消费者必须配置 isolation.level = read_committed，才能对事务外的“悬挂事务”以及“中止事务”消息进行过滤。

如果没有配置最后一步，Kafka 的事务不仅没有任何数据防重效果，反而因为引入了控制消息（Control Batch）增加了 broker 的协调负担。

为了让大家在落地 EOS 时少踩坑，我们整理了一套在 Go 语言下使用 confluent-kafka-go 的完整生产配置，包含了如何配置事务性 Producer 以及支持 read_committed 的 Consumer。

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
)

// RunTransactionalProducer 启动事务型消息生产者
func RunTransactionalProducer() {
	// 1. 初始化生产者，开启幂等和事务 ID
	p, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers":  "127.0.0.1:9092",
		"enable.idempotence": true,                      // 开启幂等性
		"transactional.id":   "accounting_tx_prod_001", // 必须指定唯一的 Transactional ID
	})
	if err != nil {
		log.Fatalf("创建生产者失败: %s", err)
	}
	defer p.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// 2. 初始化事务
	err = p.InitTransactions(ctx)
	if err != nil {
		log.Fatalf("初始化 Kafka 事务失败: %s", err)
	}

	// 3. 开启事务
	err = p.BeginTransaction()
	if err != nil {
		log.Fatalf("开启 Kafka 事务失败: %s", err)
	}

	topic := "user-wallet-events"
	msg := &kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          []byte(`{"txn_id":"TXN_20260509_9982","amount":150.00,"action":"CREDIT"}`),
	}

	// 4. 发送消息
	err = p.Produce(msg, nil)
	if err != nil {
		log.Printf("发送消息失败，准备回滚事务: %s", err)
		_ = p.AbortTransaction(context.Background())
		return
	}

	// 5. 提交事务
	err = p.CommitTransaction(context.Background())
	if err != nil {
		log.Printf("提交事务失败，进行回滚: %s", err)
		_ = p.AbortTransaction(context.Background())
		return
	}
	fmt.Println("Kafka 事务消息生产成功，并且已成功 Commit！")
}

// RunCommittedConsumer 启动只读已提交的消费者
func RunCommittedConsumer() {
	// 必须显式配置 isolation.level 为 read_committed 过滤未提交的数据
	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers":  "127.0.0.1:9092",
		"group.id":           "accounting_consumer_group",
		"auto.offset.reset":  "earliest",
		"isolation.level":    "read_committed", // 关键：只消费已 Commit 的事务消息
		"enable.auto.commit": false,            // 禁用自动提交，转为手动提交 offset
	})
	if err != nil {
		log.Fatalf("创建消费者失败: %s", err)
	}
	defer c.Close()

	err = c.SubscribeTopics([]string{"user-wallet-events"}, nil)
	if err != nil {
		log.Fatalf("订阅主题失败: %s", err)
	}

	for {
		msg, err := c.ReadMessage(3 * time.Second)
		if err != nil {
			// 读取超时或发生网络抖动，继续循环
			continue
		}

		fmt.Printf("消费到合法且已提交的事务消息: %s\n", string(msg.Value))

		// 手动提交 offset
		_, err = c.CommitMessage(msg)
		if err != nil {
			log.Printf("[WARNING] 提交 offset 失败: %v", err)
		}
	}
}

func main() {
	// 启动事务生产者示例
	RunTransactionalProducer()

	// 启动只读已提交消费者示例
	go RunCommittedConsumer()

	// 挂起主线程
	select {}
}

端到端一次性语义不仅带来了复杂度的上升，其背后的双阶段提交（Two-Phase Commit）还带来了吞吐量的显著下降。在实际业务中，对于订单记账，我们到底是用 Kafka 事务强行续命，还是应该选择在消费端设计一个极轻量级的 Redis/MySQL 幂等表（去重表）来去重处理？各位在自己的分布式架构中是如何取舍的？