Spring Boot集成Kafka：如何确保消息发送成功后再删库？

核心问题分析 #

你原本的代码逻辑存在异步风险：kafkaTemplate.send()返回ListenableFuture，属于异步操作，forEach循环会快速遍历完所有数据并触发发送请求，但此时消息可能还未被Kafka Broker确认接收，后续的deleteAll()就会直接删除数据库数据——一旦中间出现发送失败（比如Broker宕机），就会导致数据丢失。

针对你的疑问解答 #

1. Kafka发送失败会抛出异常吗？ #

默认情况下，kafkaTemplate.send()不会直接抛出异常。因为它是异步操作，发送请求只是被提交到客户端的发送队列，真正的发送结果需要通过ListenableFuture获取。如果发送失败（比如Broker宕机、网络中断），异常会被封装在Future中，只有当你主动调用future.get()或者添加回调监听时，才会触发异常抛出。

2. 是否需要额外逻辑确保所有消息发送成功后再执行删除？ #

必须要。你需要等待所有消息的发送结果被Broker确认后，再执行数据库删除操作。以下是几种可行的实现方式：

方式一：同步等待所有发送结果

通过阻塞等待每个Future完成，确保所有消息都被Broker确认：

public void syncData() throws Exception {
    List<T> data = repository.findAll();
    // 收集所有发送任务的Future
    List<ListenableFuture<SendResult<String, T>>> futures = new ArrayList<>();
    for (T value : data) {
        futures.add(kafkaTemplate.send(topicName, value));
    }
    // 等待所有Future完成，处理异常
    for (ListenableFuture<SendResult<String, T>> future : futures) {
        // 阻塞等待发送结果，超时时间可根据业务调整
        future.get(10, TimeUnit.SECONDS);
    }
    // 所有消息发送成功后，删除数据库数据
    repository.deleteAll(data);
}

方式二：异步回调+计数等待

利用ListenableFuture的回调功能，通过计数器等待所有发送完成：

public void syncData() throws InterruptedException {
    List<T> data = repository.findAll();
    if (data.isEmpty()) {
        return;
    }
    CountDownLatch latch = new CountDownLatch(data.size());
    AtomicBoolean hasError = new AtomicBoolean(false);

    for (T value : data) {
        kafkaTemplate.send(topicName, value)
                .addCallback(
                        result -> latch.countDown(),
                        ex -> {
                            hasError.set(true);
                            latch.countDown();
                            // 记录发送失败日志
                            log.error("发送消息失败: {}", value, ex);
                        }
                );
    }
    // 等待所有发送任务完成
    latch.await(30, TimeUnit.SECONDS);
    // 检查是否有发送失败
    if (hasError.get()) {
        throw new RuntimeException("存在消息发送失败，终止删除操作");
    }
    // 全部成功，删除数据
    repository.deleteAll(data);
}

方式三：事务协同（推荐）

如果你的Kafka配置支持事务（开启spring.kafka.producer.transaction-id-prefix），可以将数据库操作和Kafka发送纳入同一个事务，确保原子性：

@Transactional
public void syncData() {
    List<T> data = repository.findAll();
    // 开启Kafka事务，只有发送全部成功才会提交事务
    kafkaTemplate.executeInTransaction(template -> {
        for (T value : data) {
            template.send(topicName, value);
        }
        return null;
    });
    // 事务提交后，删除数据库数据
    repository.deleteAll(data);
}

注意：这种方式需要确保数据库和Kafka的事务能协同工作，适合对数据一致性要求较高的场景。

额外建议 #

• 考虑批量发送：如果数据量较大，使用kafkaTemplate的批量发送重载方法，减少网络交互次数，提升性能。
• 配置重试机制：在Kafka生产者配置中开启重试（spring.kafka.producer.retries），应对临时的Broker不可用情况。
• 完善日志记录：无论发送成功还是失败，都要记录详细日志，方便后续排查问题。

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