Spring Boot中实现数据库更新与Kafka消息推送的原子性事务方案

针对数据库更新完成后、Kafka消息推送前服务器崩溃导致的数据不一致问题，最优落地方案是本地消息表+定时补偿，这是业内成熟的分布式事务一致性方案，无需引入复杂中间件，易实现且稳定性高。

核心思路 #

借助数据库事务保证业务数据更新与消息记录的原子性，再通过定时任务补偿未成功发送的消息，最终实现“要么两者都成功，要么通过重试/回滚保证一致性”。

具体实现步骤 #

1. 创建本地消息表 #

在业务数据库中新增一张与业务表同库的消息记录表，字段示例：

CREATE TABLE `business_message` (
  `id` BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '主键ID',
  `business_id` VARCHAR(64) NOT NULL COMMENT '关联业务ID',
  `message_content` TEXT NOT NULL COMMENT 'Kafka消息内容',
  `status` TINYINT NOT NULL DEFAULT 0 COMMENT '状态：0-待发送，1-已发送，2-发送失败',
  `retry_count` INT NOT NULL DEFAULT 0 COMMENT '重试次数',
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间'
);

2. 修改业务逻辑 #

将数据库更新与消息记录插入放在同一个数据库事务中，确保两者原子性：

@Transactional(rollbackFor = Exception.class)
public void handleBusiness() {
    // 1. 执行数据库记录更新
    updateRecord();
    
    // 2. 插入待发送的Kafka消息到本地消息表
    BusinessMessage message = new BusinessMessage();
    message.setBusinessId("xxx"); // 关联当前业务唯一标识
    message.setMessageContent("kafka message content");
    message.setStatus(0); // 标记为待发送
    businessMessageMapper.insert(message);
    
    // 3. 尝试推送Kafka消息
    try {
        pushMessageToKafka(message.getMessageContent());
        // 推送成功，更新消息状态为已发送
        message.setStatus(1);
        businessMessageMapper.updateById(message);
    } catch (Exception e) {
        // 推送失败，不更新状态，留待定时任务补偿
        log.error("推送Kafka消息失败，业务ID：{}", message.getBusinessId(), e);
    }
}

3. 定时补偿任务 #

编写定时任务，定期扫描本地消息表中待发送或重试中的消息，重新推送：

@Component
public class MessageCompensateTask {
    @Autowired
    private BusinessMessageMapper businessMessageMapper;
    @Autowired
    private KafkaProducer kafkaProducer;
    
    // 每5分钟执行一次，可根据业务调整间隔
    @Scheduled(cron = "0 */5 * * * ?")
    public void compensateUnsentMessages() {
        // 查询状态为待发送、且创建时间超过1分钟的消息（避免刚插入就重试）
        List<BusinessMessage> messages = businessMessageMapper.selectUnsentMessages();
        
        for (BusinessMessage message : messages) {
            if (message.getRetryCount() >= 3) { // 最大重试次数，可配置
                // 超过重试次数，标记为发送失败并触发告警
                message.setStatus(2);
                businessMessageMapper.updateById(message);
                log.error("消息重试次数耗尽，业务ID：{}", message.getBusinessId());
                // 此处可添加告警逻辑，如通知运维人员
                continue;
            }
            
            try {
                kafkaProducer.pushMessageToKafka(message.getMessageContent());
                // 推送成功，更新状态和重试次数
                message.setStatus(1);
                message.setRetryCount(message.getRetryCount() + 1);
                businessMessageMapper.updateById(message);
            } catch (Exception e) {
                // 推送失败，仅更新重试次数
                message.setRetryCount(message.getRetryCount() + 1);
                businessMessageMapper.updateById(message);
                log.error("补偿推送Kafka消息失败，业务ID：{}，重试次数：{}", message.getBusinessId(), message.getRetryCount(), e);
            }
        }
    }
}

4. 处理Pod崩溃场景 #

• 如果Pod在数据库更新和消息插入成功后、推送消息前崩溃：重启后定时任务会扫描到这条待发送的消息，自动重试推送，保证消息最终发送成功。
• 如果Pod在推送消息成功后、未更新消息状态就崩溃：重启后定时任务会再次推送，此时需要消费者端实现幂等性（比如通过业务ID或消息ID去重），避免重复消费导致业务异常。

注意事项 #

• 本地消息表必须与业务表在同一个数据库，才能利用数据库事务保证原子性。
• 消费者端必须实现幂等性，这是分布式场景下的必备要求。
• 定时任务的执行间隔、最大重试次数需根据业务场景调整，避免对系统造成过大压力。

内容的提问来源于stack exchange，提问作者Aki T