Spring Kafka与数据库/MQ交互场景下的事务管理问题咨询

你当前所有@Transactional注解都绑定了KafkaTransactionManager，该事务管理器仅负责管控Kafka侧的逻辑：包括消费偏移量提交、生产者消息的提交/回滚。而JPA的数据库操作并未被纳入该事务的管控范围，因此数据库提交失败时，不会触发Kafka事务的回滚，最终导致系统数据不一致。

方案1：发件箱模式（高一致性场景优先推荐） #

这是目前分布式事务中「消费-数据库操作-生产消息」场景的官方最佳实践，可靠性最高，只要数据库不丢数据就不会出现消息丢失或不一致问题。

实现步骤： #

1. 新增数据库发件箱表，用于存储待发送到Kafka的消息：

CREATE TABLE outbox (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  target_topic VARCHAR(255) NOT NULL,
  msg_key VARCHAR(255),
  msg_content TEXT NOT NULL,
  send_status TINYINT NOT NULL DEFAULT 0 COMMENT '0=待发送 1=已发送 2=发送失败',
  create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

2. 调整消费逻辑，将业务操作和发件箱写入绑定到同一个数据库本地事务：

@KafkaHandler
@Transactional(transactionManager = "jpaTransactionManager")
public void receiveCreationMessage(String event, Acknowledgment ack){
  // 先做幂等校验，避免重复消费
  if (idempotentService.checkIfConsumed(event.getUniqueId())) {
    ack.acknowledge();
    return;
  }
  // 执行业务数据库操作
  dao.saveDraft(event);
  // 待发送的Kafka消息写入发件箱
  OutboxEntity outbox = new OutboxEntity();
  outbox.setTargetTopic("Topic B");
  outbox.setMsgKey(event.getMsgKey());
  outbox.setMsgContent(convertedEvent);
  outboxDao.save(outbox);
  // 标记消息已消费，保存幂等记录
  idempotentService.markConsumed(event.getUniqueId());
  // 提交消费偏移量
  ack.acknowledge();
}

3. 新增独立的异步定时任务，负责投递发件箱中的消息：

@Scheduled(fixedDelay = 100)
@Transactional
public void deliverOutboxMsg() {
  // 拉取待发送消息，加行锁避免多实例并发重复投递
  List<OutboxEntity> toSendList = outboxDao.findTop100BySendStatusOrderByCreateTimeAsc(0, LockModeType.PESSIMISTIC_WRITE);
  for (OutboxEntity outbox : toSendList) {
    try {
      kafkaTemplate.send(outbox.getTargetTopic(), outbox.getMsgKey(), outbox.getMsgContent()).get();
      // 标记为已发送
      outbox.setSendStatus(1);
      outboxDao.save(outbox);
    } catch (Exception e) {
      // 标记为发送失败，后续可通过告警人工介入或重试逻辑处理
      outbox.setSendStatus(2);
      outboxDao.save(outbox);
      log.error("发件箱消息投递失败，id:{}", outbox.getId(), e);
    }
  }
}

方案2：事务同步回调（轻量方案，适合一致性要求一般的场景） #

通过Spring提供的事务同步机制，将Kafka消息发送和偏移量提交绑定到数据库事务成功提交之后执行，实现轻量级的一致性保障。

实现代码： #

@KafkaHandler
@Transactional(transactionManager = "jpaTransactionManager")
public void receiveCreationMessage(String event, Acknowledgment ack){
  // 幂等校验
  if (idempotentService.checkIfConsumed(event.getUniqueId())) {
    ack.acknowledge();
    return;
  }
  // 执行数据库业务操作
  dao.saveDraft(event);
  // 注册事务同步回调
  TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
    @Override
    public void afterCommit() {
      // 数据库事务提交成功后才发送Kafka消息
      try {
        sendMQAdapterKafkaEvent(initiateReqMq, convertedEvent);
        ack.acknowledge();
        idempotentService.markConsumed(event.getUniqueId());
      } catch (Exception e) {
        // 消息发送失败时存入重试表，后续异步重试，避免消息丢失
        retryService.saveRetryTask(event);
        log.error("Kafka消息发送失败，消息id:{}", event.getUniqueId(), e);
      }
    }
  });
}

注意事项： #

该方案存在极小概率的消息丢失风险：如果应用在数据库提交成功后、Kafka消息发送前宕机，消息就会丢失，因此需要配合重试机制和下游消费的幂等校验使用。

• 对数据一致性要求高的核心业务场景，优先选择发件箱模式
• 非核心、可接受极低概率不一致的场景，可选择更轻量的事务同步回调方案

内容的提问来源于stack exchange，提问作者Bansi Derashri