如何实现Kafka消息可靠发送并保持数据库数据一致性，采用异步处理和事务模式？

本文共计2272个文字，预计阅读时间需要10分钟。

在分布式系统中，将数据从关系型数据库同步到消息队列（如Kafka）是一个常见场景。为确保数据在成功发送到Kafka后才从数据库中删除，可以采取以下步骤：

1. 异步发送与回调机制

kafkaTemplate.send方法返回的ListenableFuture允许我们注册回调函数，以在消息发送成功或失败时执行特定逻辑。这是处理异步操作结果的基本方式。

1.1 问题分析

直接在kafkaTemplate.send之后删除数据库数据是危险的，因为ListenableFuture的特性意味着发送操作尚未完成。如果Kafka Broker在消息发送过程中宕机或网络出现问题，消息可能未能成功发送，而数据库中的原始数据却已被删除，导致数据丢失。

// 原始的、存在风险的代码示例 public void syncDataRisky() { List<YourDataModel> data = repository.findAll(); data.forEach(value -> kafkaTemplate.send(topicName, value.getKey(), value)); // 风险点：此行代码可能在所有Kafka消息真正发送成功前执行 repository.deleteAll(data); }

1.2 解决方案：利用ListenableFutureCallback

正确的做法是为每个ListenableFuture添加一个回调，仅在消息成功发送后才执行数据库删除操作。