跳至主要內容

Kafka Exactly-Once语义：幂等Producer配置与事务API的完整实现

老张大约 9 分钟

Kafka Exactly-Once语义：幂等Producer配置与事务API的完整实现

适读人群：需要理解Kafka EOS原理，在流处理场景实现精确一次语义的Java工程师 | 阅读时长：约17分钟

开篇故事

在一个实时数据聚合系统里，我们用Kafka Streams做用户行为统计：消费原始行为事件Topic，聚合后写回统计结果Topic，下游消费统计结果做报表展示。

系统运行得不错，直到某天压测时，我们故意随机Kill一些消费者实例来模拟故障。结果发现：报表中同一个用户在同一时间窗口的点击量，可能被计算了两次——因为消费者故障后从上一个checkpoint恢复，中间已经计算过的数据又重新计算了一遍，结果写了两次。

这就是Kafka的"至少一次"（At-Least-Once）语义的代价：消费者可能重复处理消息，下游可能收到重复数据。

Kafka 0.11版本引入了两个重要特性来解决这个问题：幂等Producer（解决Producer重试导致的重复发送）和事务API（解决消费-处理-生产的原子性问题）。今天全面讲透。

一、Exactly-Once的三个层次

要理解Kafka的EOS，需要分清三个不同层次的"精确一次"：

应用层幂等（第435期讲的Redis NX、DB唯一索引）是业务层面的幂等，是Kafka EOS无法覆盖的消费端业务逻辑幂等。两者不是竞争关系，而是不同层次的保障。

二、幂等Producer原理

2.1 为什么Producer会重复发送

即使配置了acks=all，也可能出现重复发送：

2.2 幂等Producer的去重机制

幂等Producer（enable.idempotence=true）通过ProducerID（PID）+ 序列号（SequenceNumber）实现去重：

  • Broker为每个Producer分配唯一的PID
  • Producer对每个TopicPartition维护单调递增的序列号
  • Broker收到消息时检查序列号，如果已存在（重复），直接返回成功但不重复写入
/**
 * 幂等Producer配置
 * 注意：幂等只在单个会话（Producer生命周期内）有效
 * 如果Producer重启，PID会变，之前的去重状态丢失
 */
@Bean
public ProducerFactory<String, String> idempotentProducerFactory() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
        "kafka1:9092,kafka2:9092,kafka3:9092");
    
    // 开启幂等：自动设置acks=all, retries=MAX_INT, max.in.flight.requests.per.connection<=5
    configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    
    // 以下配置由幂等自动保证，可以不配但不能与幂等冲突
    // configs.put(ProducerConfig.ACKS_CONFIG, "all");          // 幂等要求
    // configs.put(ProducerConfig.RETRIES_CONFIG, MAX_VALUE);   // 幂等要求
    // configs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); // 幂等要求
    
    configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    
    return new DefaultKafkaProducerFactory<>(configs);
}

幂等的局限性：

  1. 只在单次Producer会话内有效，重启后PID变化，幂等状态清零
  2. 只保证单个TopicPartition内的去重，不保证跨Partition的原子性
  3. 网络分区恢复后，无法保证跨会话的去重

三、Kafka事务API原理

3.1 事务的保障范围

Kafka事务解决的是：原子性地将一组消息写入多个TopicPartition，以及原子性地提交消费offset。

要么全部写入成功，要么全部失败。消费者只能看到已提交的事务中的消息（需要设置isolation.level=read_committed）。

3.2 事务ID与事务协调者

3.3 transactional.id 的作用

transactional.id是事务Producer的全局唯一标识符。有了它，Kafka才能实现"跨会话的精确一次"：

  • 相同transactional.id的新Producer启动时，会找到上次的未完成事务并恢复/回滚
  • Broker通过transactional.id隔离不同Producer实例，防止"僵尸写入"（旧Producer实例的写入不被新实例看到）

四、完整Java实现

4.1 事务Producer实现

/**
 * Kafka事务Producer配置
 * 场景：Kafka Streams风格的 消费->处理->生产 原子操作
 */
@Configuration
public class KafkaTransactionConfig {

    @Bean
    public ProducerFactory<String, String> transactionalProducerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
            "kafka1:9092,kafka2:9092,kafka3:9092");
        
        // 事务ID：必须全局唯一，且在重启后保持不变
        // 在K8s环境中，可以用StatefulSet的Pod名称
        String podName = System.getenv("POD_NAME");
        String transactionalId = "order-agg-producer-" + 
            (podName != null ? podName : "local");
        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);
        
        // 事务超时：事务在Broker端的最大存活时间（默认60秒）
        configs.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000);
        
        // 幂等（事务Producer自动开启幂等）
        configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        
        DefaultKafkaProducerFactory<String, String> factory = 
            new DefaultKafkaProducerFactory<>(configs);
        
        // Spring Kafka的事务支持：配置transactionIdPrefix后，
        // Spring会自动管理事务Producer的生命周期
        factory.setTransactionIdPrefix("tx-order-agg-");
        
        return factory;
    }

    @Bean
    public KafkaTransactionManager<String, String> kafkaTransactionManager(
            ProducerFactory<String, String> pf) {
        return new KafkaTransactionManager<>(pf);
    }
}

4.2 消费-处理-生产的原子性实现

/**
 * 端到端Exactly-Once实现
 * 原子操作：消费原始事件 -> 计算统计数据 -> 写入结果Topic + 提交offset
 */
@Service
@Slf4j
public class ExactlyOnceProcessor {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final StatisticsCalculator calculator;

    public ExactlyOnceProcessor(KafkaTemplate<String, String> kafkaTemplate,
                                 StatisticsCalculator calculator) {
        this.kafkaTemplate = kafkaTemplate;
        this.calculator = calculator;
    }

    /**
     * 事务性处理：消费事件 -> 更新统计 -> 发送结果
     * 使用Spring的@Transactional(transactionManager = "kafkaTransactionManager")
     */
    @KafkaListener(
        topics = "raw-events",
        groupId = "agg-consumer-group",
        containerFactory = "exactlyOnceContainerFactory"
    )
    @Transactional(transactionManager = "kafkaTransactionManager")
    public void process(List<ConsumerRecord<String, String>> records,
                        Acknowledgment acknowledgment) {
        
        // 1. 计算聚合结果
        Map<String, Long> aggregated = records.stream()
            .map(r -> parseEvent(r.value()))
            .filter(Objects::nonNull)
            .collect(Collectors.groupingBy(
                UserEvent::getUserId,
                Collectors.counting()
            ));
        
        // 2. 在同一个事务中发送结果到结果Topic
        aggregated.forEach((userId, count) -> {
            UserStats stats = new UserStats();
            stats.setUserId(userId);
            stats.setEventCount(count);
            stats.setWindowStart(getWindowStart(records));
            
            // 这条发送和后续的offset提交在同一个事务中
            kafkaTemplate.send("user-stats-topic", userId, 
                JsonUtil.toJson(stats));
        });
        
        // 3. 提交offset（和上面的发送在同一个事务中！）
        // Spring Kafka会自动在事务提交前处理offset
        acknowledgment.acknowledge();
        
        log.info("事务处理完成: records={}, users={}", 
            records.size(), aggregated.size());
    }
    
    private UserEvent parseEvent(String value) {
        try {
            return JsonUtil.fromJson(value, UserEvent.class);
        } catch (Exception e) {
            return null;
        }
    }
    
    private LocalDateTime getWindowStart(List<ConsumerRecord<String, String>> records) {
        // 获取时间窗口开始时间
        return LocalDateTime.now().withMinute(0).withSecond(0);
    }
}

/**
 * Exactly-Once消费者容器工厂
 * 关键：isolation.level=read_committed
 */
@Bean("exactlyOnceContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> exactlyOnceContainerFactory() {
    Map<String, Object> consumerProps = new HashMap<>();
    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
        "kafka1:9092,kafka2:9092,kafka3:9092");
    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "agg-consumer-group");
    
    // 关键配置：只读已提交事务的消息
    // read_uncommitted（默认）：读取所有消息，包括未提交事务的消息
    // read_committed：只读取已提交事务的消息（和生产者事务配合使用）
    consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
    
    consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
        StringDeserializer.class);
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
        StringDeserializer.class);
    
    ConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
    
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf);
    factory.setBatchListener(true);
    
    ContainerProperties props = factory.getContainerProperties();
    props.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    
    // 配置KafkaTransactionManager，Spring会自动管理事务
    // props.setKafkaAwareTransactionManager(kafkaTransactionManager);
    
    return factory;
}

4.3 原生API直接使用事务

/**
 * 原生Kafka事务API（不使用Spring封装）
 * 更直观地展示事务的完整流程
 */
@Component
@Slf4j
public class NativeTransactionExample {

    private Producer<String, String> transactionalProducer;

    @PostConstruct
    public void init() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "native-tx-producer-001");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        
        transactionalProducer = new KafkaProducer<>(props);
        
        // 初始化事务（必须调用，且只调用一次）
        transactionalProducer.initTransactions();
    }

    /**
     * 原子性地写入多个Partition
     */
    public void atomicWrite(String topic1, String topic2, String key, String value) {
        try {
            transactionalProducer.beginTransaction();
            
            transactionalProducer.send(new ProducerRecord<>(topic1, key, value));
            transactionalProducer.send(new ProducerRecord<>(topic2, key, value));
            
            transactionalProducer.commitTransaction();
            log.info("事务提交成功");
            
        } catch (ProducerFencedException e) {
            // 相同transactional.id的新实例启动，当前实例被"fence"
            // 必须关闭当前Producer，不能再使用
            log.error("Producer被fence，关闭: {}", e.getMessage());
            transactionalProducer.close();
            
        } catch (KafkaException e) {
            // 其他Kafka异常，中止事务
            log.error("事务异常，中止: {}", e.getMessage());
            transactionalProducer.abortTransaction();
        }
    }
}

四、踩坑实录

坑1：transactional.id在K8s下的重复

在K8s多实例部署时，如果所有Pod使用相同的transactional.id（比如固定字符串），启动第二个Pod时，Broker会认为新Pod的事务与旧Pod冲突，旧Pod会被"fence"（ProducerFencedException），无法再发送消息。

解决方案：用K8s StatefulSet，每个Pod有固定编号，用编号作为transactional.id的一部分：

String transactionalId = "order-agg-" + System.getenv("HOSTNAME"); // Pod-0, Pod-1...

坑2：事务超时导致消息可见性延迟

事务超时时间设得太长（默认60秒），某个事务Producer因为某些原因迟迟没有提交事务，下游消费者（read_committed模式）无法消费这个事务之后的任何消息，造成消费"假卡死"。

解决方案：合理设置transaction.timeout.ms（不超过30秒），并监控生产者的事务提交情况。

坑3：read_committed消费者延迟更高

isolation.level=read_committed的消费者需要等待事务提交标记（COMMIT/ABORT）才能消费，相比read_uncommitted会有额外延迟（通常几毫秒到几十毫秒，取决于事务大小）。

在对延迟极度敏感的场景（如实时竞价广告），这个额外延迟可能不可接受。

权衡：只在真正需要EOS保障的场景用事务API，对延迟敏感的场景用应用层幂等代替。

坑4：事务和压缩日志的冲突

事务Producer写入的消息包含特殊的事务标记消息，Log Compaction会把这些标记也纳入压缩，可能导致事务标记丢失，影响消费者的事务状态判断。

解决方案：启用事务的Topic不要开启Log Compaction（cleanup.policy=compact），两者不兼容。

五、EOS性能代价

语义配置相比At-Most-Once吞吐降幅延迟增加
At-Most-Onceacks=0基准基准(~1ms)
At-Least-Onceacks=1, enable.idempotence=false-40%+5ms
At-Least-Onceacks=all-60%+15ms
Exactly-Once幂等+事务-70%+20-50ms

EOS的代价确实不小，但对于需要精确计数、金额计算的场景，这个代价是值得的。对于可以接受重复的场景，用应用层幂等就够了，不需要引入Kafka事务的复杂性。

下一篇（第442期）讲RocketMQ顺序消息，全局顺序和分区顺序的实现原理，以及为了顺序消息要付出的性能代价。