Kafka消息发送与数据库数据删除的事务一致性保障

public void syncData() {
    List<T> data = repository.findAll();
    // 潜在风险：forEach循环可能在所有消息实际发送前完成
    data.forEach(value -> kafkaTemplate.send(topicName, value));
    // 如果上述发送失败，这些数据可能被错误删除
    repository.deleteAll(data);
}

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public class DataSyncService<T> {

    private final KafkaTemplate<String, T> kafkaTemplate;
    private final YourRepository<T> repository; // 假设有一个数据仓库接口
    private final String topicName;

    public DataSyncService(KafkaTemplate<String, T> kafkaTemplate, YourRepository<T> repository, String topicName) {
        this.kafkaTemplate = kafkaTemplate;
        this.repository = repository;
        this.topicName = topicName;
    }

    public void syncDataSafely() {
        List<T> dataToSync = repository.findAllPendingData(); // 假设只查询待同步数据
        if (dataToSync.isEmpty()) {
            return;
        }

        // 使用计数器来跟踪所有消息的发送状态
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failureCount = new AtomicInteger(0);
        AtomicInteger totalSends = new AtomicInteger(dataToSync.size());

        for (T item : dataToSync) {
            ListenableFuture<SendResult<String, T>> future = kafkaTemplate.send(topicName, item);
            future.addCallback(new ListenableFutureCallback<SendResult<String, T>>() {
                @Override
                public void onSuccess(SendResult<String, T> result) {
                    System.out.println("Message sent successfully: " + result.getProducerRecord().value());
                    // 只有当消息成功发送后，才考虑从数据库中删除或标记该数据
                    repository.delete(item); // 或者更新状态为“已发送”
                    successCount.incrementAndGet();
                    checkCompletion(totalSends.get(), successCount.get(), failureCount.get());
                }

                @Override
                public void onFailure(Throwable ex) {
                    System.err.println("Failed to send message: " + item + ", Error: " + ex.getMessage());
                    // 记录错误，可能需要重试或人工干预
                    failureCount.incrementAndGet();
                    checkCompletion(totalSends.get(), successCount.get(), failureCount.get());
                }
            });
        }
        // 注意：这里不能直接进行 deleteAll(dataToSync)，因为回调是异步的
        // 实际的删除逻辑应该在所有回调都完成后触发，或者每个成功回调单独处理
    }

    // 辅助方法：检查所有消息是否都已处理
    private void checkCompletion(int total, int success, int failure) {
        if (success + failure == total) {
            System.out.println("All messages processed. Success: " + success + ", Failure: " + failure);
            // 这里可以触发一个全局的完成事件，例如清理所有已发送的数据
            // 但更推荐在每个 onSuccess 中处理单个数据的删除或标记
        }
    }
}

// 1. 业务逻辑与Outbox记录在同一事务中
@Transactional
public void createOrderAndPublishEvent(Order order) {
    // 保存业务数据
    orderRepository.save(order);

    // 创建Outbox记录
    OutboxMessage outboxMessage = new OutboxMessage();
    outboxMessage.setTopic("order_events");
    outboxMessage.setKey(order.getId().toString());
    outboxMessage.setPayload(objectMapper.writeValueAsString(order)); // 将Order对象序列化
    outboxMessage.setStatus("PENDING"); // 待发送状态
    outboxMessageRepository.save(outboxMessage);

    // 此时，数据库中订单和Outbox记录要么都存在，要么都不存在
}

// 2. 独立的Outbox发送服务（通常是定时任务或Kafka Connect）
@Scheduled(fixedRate = 5000) // 每5秒检查一次
public void processOutbox() {
    List<OutboxMessage> pendingMessages = outboxMessageRepository.findByStatus("PENDING");
    for (OutboxMessage message : pendingMessages) {
        try {
            kafkaTemplate.send(message.getTopic(), message.getKey(), message.getPayload()).addCallback(
                result -> {
                    // 成功发送，更新Outbox记录状态或删除
                    message.setStatus("SENT");
                    outboxMessageRepository.save(message);
                },
                ex -> {
                    System.err.println("Failed to send outbox message: " + message.getId() + ", Error: " + ex.getMessage());
                    // 记录错误，可能需要重试策略或标记为FAILED
                    message.setStatus("FAILED"); // 或增加重试次数
                    outboxMessageRepository.save(message);
                }
            );
        } catch (Exception e) {
            System.err.println("Error processing outbox message: " + message.getId() + ", Error: " + e.getMessage());
            // 捕获同步发送时的异常
            message.setStatus("FAILED");
            outboxMessageRepository.save(message);
        }
    }
}