Apache Kafka与SpringBoot的精确一次生产与消费问题咨询

默认幂等性状态 #

• KafkaTemplate生产者：默认不开启幂等性。Kafka原生生产者的幂等性依赖enable.idempotence参数，Spring Kafka的KafkaTemplate默认未启用该配置，因此网络重试等场景下可能出现重复消息。
• @KafkaListener消费者：默认不具备幂等性。默认采用自动提交offset机制，若消费完成后offset未提交就宕机，重启后会重复消费；若offset提交成功但邮件发送失败，则会丢失通知，无法满足Exactly once要求。

配置KafkaTemplate实现生产者幂等性 #

开启生产者幂等性需要配置以下核心参数，确保Broker能基于生产者ID和消息序列号去重：

1. YAML配置文件方式 #

spring:
  kafka:
    producer:
      # 开启幂等性
      enable-idempotence: true
      # 必须设置为all，幂等性依赖Broker的全部确认
      acks: all
      # 自定义生产者ID，用于Broker识别唯一生产者
      client-id: microservice-a-producer
      # 序列化配置（根据你的Event类型调整）
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

2. Java配置类方式 #

如果需要自定义KafkaTemplate，可通过配置类实现：

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, Event> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // 开启幂等性
        configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        // 必须设置acks为all
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");
        configProps.put(ProducerConfig.CLIENT_ID_CONFIG, "microservice-a-producer");
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, Event> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

配置@KafkaListener实现消费者幂等性与Exactly Once #

消费者侧的Exactly once需要结合手动offset提交和业务层幂等校验，或通过Kafka事务实现更严谨的端到端一致性。

1. 基础幂等方案（手动提交offset） #

先关闭自动提交，开启手动提交模式：

spring:
  kafka:
    consumer:
      enable-auto-commit: false
      group-id: microservice-b-consumer
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        # 指定Event类的全路径，用于JSON反序列化
        spring.json.value.default.type: com.yourpackage.Event
        spring.json.trusted.packages: com.yourpackage
    listener:
      # 手动立即提交offset
      ack-mode: MANUAL_IMMEDIATE

在消费逻辑中加入幂等校验，确保只有未处理过的消息才会执行邮件发送：

@KafkaListener(topics = "${kafka.topic.event}")
public void consumeEvent(ConsumerRecord<String, Event> record, Acknowledgment acknowledgment) {
    Event event = record.value();
    String stepId = event.getStepId();

    // 1. 检查该步骤是否已发送过通知（依赖数据库等持久化存储）
    if (notificationService.isStepNotified(stepId)) {
        acknowledgment.acknowledge();
        return;
    }

    try {
        // 2. 执行邮件发送
        emailService.sendCompletionNotice(event);
        // 3. 持久化已通知状态（与邮件发送原子性，可通过数据库事务保证）
        notificationService.markStepAsNotified(stepId);
        // 4. 提交offset，确认消息已处理完成
        acknowledgment.acknowledge();
    } catch (Exception e) {
        log.error("Failed to process step {} notification", stepId, e);
        // 不提交offset，Kafka会重新投递该消息
    }
}

2. Kafka事务端到端Exactly Once（高一致性场景） #

如果需要严格的端到端Exactly once，可结合生产者事务和消费者事务：

生产者事务配置

在生产者配置中添加事务前缀，并配置事务管理器：

spring:
  kafka:
    producer:
      enable-idempotence: true
      acks: all
      client-id: microservice-a-producer
      # 事务ID前缀，用于生成唯一事务ID
      transaction-id-prefix: tx-ms-a-
    # 绑定Kafka事务管理器
    transaction-manager: kafkaTransactionManager

Java配置类添加事务管理器：

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

在微服务A中，用事务包裹步骤执行和消息发送，确保步骤执行成功才提交消息：

@Service
public class EventProducer {
    private final KafkaTemplate<String, Event> kafkaTemplate;
    private final StepService stepService;

    @Autowired
    public EventProducer(NewTopic topic, KafkaTemplate<String, Event> kafkaTemplate, StepService stepService) {
        this.kafkaTemplate = kafkaTemplate;
        this.stepService = stepService;
    }

    @Transactional(transactionManager = "kafkaTransactionManager")
    public void sendStepCompletedEvent(Event event) {
        // 执行步骤业务逻辑（如数据库操作）
        stepService.finishStep(event.getStepId());
        // 事务内发送消息，只有业务逻辑成功才会提交事务
        kafkaTemplate.send("event-topic", event);
    }
}

消费者事务配置

消费者开启事务，并设置隔离级别为read_committed，确保只消费已提交的事务消息：

spring:
  kafka:
    consumer:
      enable-auto-commit: false
      group-id: microservice-b-consumer
      properties:
        isolation.level: read_committed
        spring.json.value.default.type: com.yourpackage.Event
        spring.json.trusted.packages: com.yourpackage
    listener:
      ack-mode: RECORD
      # 开启事务性消费
      transactional: true

消费逻辑用事务包裹，确保邮件发送和offset提交原子性：

@KafkaListener(topics = "${kafka.topic.event}")
@Transactional
public void consumeEvent(Event event) {
    String stepId = event.getStepId();
    if (notificationService.isStepNotified(stepId)) {
        return;
    }
    emailService.sendCompletionNotice(event);
    notificationService.markStepAsNotified(stepId);
    // 事务提交时自动提交offset
}

关键注意事项 #

• 生产者幂等性仅保证同一生产者不会重复写入，多生产者场景仍需消费者端幂等校验。
• 消费者端的幂等校验是必选项，即使生产者无重复，Kafka的重试机制仍可能导致重复消费。
• Exactly once语义需要生产者幂等、消费者幂等、事务/手动提交三者配合，缺一不可。

内容的提问来源于stack exchange，提问作者Spinozist Scientist