Spring Boot实现MongoDB持久化与Kafka发送的原子性方案咨询

以下是几种可落地的实现方案，按一致性强度和落地成本排序：

方案一：本地事务+Kafka重试机制（强一致性，单节点MongoDB适用） #

MongoDB 4.0+单节点支持ACID事务，可将MongoDB写入放在本地事务内，事务提交成功后再发送Kafka消息，同时通过Kafka生产者重试机制保证消息必达。

具体实现： #

1. 开启MongoDB事务支持

@Configuration
public class MongoConfig extends AbstractMongoClientConfiguration {
    @Override
    protected String getDatabaseName() {
        return "your_db_name";
    }

    @Bean
    public MongoTransactionManager transactionManager(MongoDatabaseFactory dbFactory) {
        return new MongoTransactionManager(dbFactory);
    }
}

2. 业务方法绑定事务，提交后发送Kafka消息

@Service
public class BusinessService {
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional(transactionManager = "mongoTransactionManager")
    public void saveAndSend(Object data) {
        // 事务内执行MongoDB持久化
        mongoTemplate.save(data);
        // 事务提交成功后发送Kafka消息，添加回调处理结果
        kafkaTemplate.send("your_topic", data).addCallback(
                success -> log.info("Kafka消息发送成功: {}", success.getRecordMetadata()),
                failure -> log.error("Kafka消息发送失败，将触发重试: {}", failure.getMessage())
        );
    }
}

3. 配置Kafka生产者重试参数

spring.kafka.producer.retries=3
spring.kafka.producer.retry-backoff-ms=1000
spring.kafka.producer.acks=all
spring.kafka.producer.enable-idempotence=true

方案二：MongoDB变更流（Change Streams）异步触发（最终一致性，集群环境适用） #

针对MongoDB副本集/分片集群，可利用Change Streams监听数据变更，一旦检测到插入操作，自动发送Kafka消息，确保数据写入MongoDB后必触发消息发送。

具体实现： #

@Component
public class MongoChangeStreamListener {
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @PostConstruct
    public void startListening() {
        mongoTemplate.changeStream(ChangeStreamOptions.empty(), Object.class, "your_collection")
                .listen()
                .subscribe(changeStreamDoc -> {
                    // 仅处理插入类型的变更
                    if (ChangeStreamOperationType.INSERT.equals(changeStreamDoc.getOperationType())) {
                        Object data = changeStreamDoc.getFullDocument();
                        kafkaTemplate.send("your_topic", data);
                    }
                });
    }
}

注意：需确保MongoDB集群已启用副本集，Change Streams依赖副本集的oplog实现。

方案三：可靠消息最终一致性（本地消息表+定时补偿） #

通过在MongoDB中维护消息日志表，将业务数据写入与消息记录放在同一事务，再通过定时任务扫描未发送的消息进行重试，保证最终一致性。

具体步骤： #

1. 定义消息日志实体

@Document(collection = "message_log")
public class MessageLog {
    @Id
    private String id;
    private String topic;
    private Object content;
    private Integer status; // 0:待发送, 1:发送成功, 2:发送失败
    private LocalDateTime createTime;
    private LocalDateTime updateTime;
    // getter/setter
}

2. 业务方法绑定事务，同时写入业务数据和消息日志

@Transactional(transactionManager = "mongoTransactionManager")
public void saveAndRecordMessage(Object businessData, MessageLog messageLog) {
    mongoTemplate.save(businessData);
    mongoTemplate.save(messageLog);
}

3. 定时任务重试发送未成功的消息

@Component
public class MessageRetryTask {
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Scheduled(fixedRate = 5000)
    public void retrySendMessages() {
        Query query = new Query(Criteria.where("status").is(0));
        List<MessageLog> messageLogs = mongoTemplate.find(query, MessageLog.class);
        for (MessageLog log : messageLogs) {
            try {
                kafkaTemplate.send(log.getTopic(), log.getContent()).get();
                log.setStatus(1);
            } catch (Exception e) {
                log.setStatus(2);
                log.setUpdateTime(LocalDateTime.now());
            }
            mongoTemplate.save(log);
        }
    }
}

方案四：分布式事务（Seata等框架） #

若系统已引入分布式事务框架（如Seata），可通过AT模式管理MongoDB与Kafka的分布式事务。需注意Seata对MongoDB的支持需额外配置，且Kafka需适配Seata事务分支逻辑。

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