确保Kafka消息可靠发送与数据库数据一致性：异步处理与事务模式

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

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList; // 用于线程安全地收集成功ID

@Service
public class DataSyncService {

    private final KafkaTemplate<String, YourDataModel> kafkaTemplate;
    private final YourRepository repository; // 假设YourRepository处理YourDataModel

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

    public void syncDataWithKafkaCallbacks() {
        // 假设这里获取的是需要同步的数据
        List<YourDataModel> dataToSync = repository.findAllPendingSync();
        if (dataToSync.isEmpty()) {
            return;
        }

        // 用于收集成功发送的消息对应的ID，以便后续批量删除
        List<Long> successfullySentIds = new CopyOnWriteArrayList<>();

        for (YourDataModel data : dataToSync) {
            ListenableFuture<SendResult<String, YourDataModel>> future =
                    kafkaTemplate.send("your-topic", data.getKey(), data);

            future.addCallback(new ListenableFutureCallback<SendResult<String, YourDataModel>>() {
                @Override
                public void onSuccess(SendResult<String, YourDataModel> result) {
                    System.out.println("消息 " + data.getId() + " 成功发送到主题 " + result.getRecordMetadata().topic());
                    // 仅在Kafka发送成功后，将数据ID加入待删除列表
                    successfullySentIds.add(data.getId());
                }

                @Override
                public void onFailure(Throwable ex) {
                    System.err.println("消息 " + data.getId() + " 发送失败. 错误: " + ex.getMessage());
                    // 处理发送失败：例如记录日志、将数据标记为待重试、或发送告警
                    // 注意：在此处不应删除数据库数据
                }
            });
        }

        // 注意：在实际生产环境中，不应立即在此处进行批量删除
        // 因为上述循环中的回调是异步的，此处的代码会立即执行。
        // 正确的做法是：等待所有Future完成，或者通过一个独立的定时任务来处理successfullySentIds。
        // 对于本教程的演示目的，我们假设在所有回调处理完成后，再进行批量删除。
        // 实际应用中，可能需要使用CountDownLatch或CompletableFuture.allOf()来等待所有Future完成。
        // 或者更推荐Outbox模式。
    }

    /**
     * 假设这是一个独立的清理方法，可以在所有发送操作（及其回调）完成后调用，
     * 或者由一个独立的消费者服务消费成功事件后触发。
     * 实际应用中，需要更复杂的机制来确保所有Future都已完成。
     */
    public void cleanupSentData(List<Long> idsToDelete) {
        if (!idsToDelete.isEmpty()) {
            repository.deleteAllById(idsToDelete); // 批量删除
            System.out.println("成功删除数据库中已发送的 " + idsToDelete.size() + " 条数据。");
        }
    }
}

spring:
  kafka:
    producer:
      bootstrap-servers: localhost:9092 # Kafka Broker地址
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 或其他序列化器
      acks: all # 确保所有同步副本确认消息，提供最高可靠性
      retries: 3 # 生产者在发送失败后重试的次数
      retry-backoff-ms: 100 # 重试间隔
      # batch-size: 16384 # 批量发送大小，有助于提高吞吐量
      # linger.ms: 1 # 批量发送等待时间，与batch-size协同工作

// 假设这是您的业务实体
@Entity
public class Order {
    @Id
    private Long id;
    private String status;
    // ... 其他业务字段
}

// Outbox表实体
@Entity
public class OutboxMessage {
    @Id
    private Long id;
    private String topic;
    private String key;
    private String payload; // JSON序列化后的消息内容
    private LocalDateTime createdAt;
    private String status; // PENDING, SENT, FAILED
}

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxMessageRepository outboxMessageRepository;

    public OrderService(OrderRepository orderRepository, OutboxMessageRepository outboxMessageRepository) {
        this.orderRepository = orderRepository;
        this.outboxMessageRepository = outboxMessageRepository;
    }

    @Transactional // 确保数据库操作的原子性
    public void createOrderAndPublishEvent(Order order) {
        orderRepository.save(order); // 保存业务数据

        OutboxMessage outboxMessage = new OutboxMessage();
        outboxMessage.setTopic("order-events");
        outboxMessage.setKey(order.getId().toString());
        outboxMessage.setPayload(serializeOrderToJson(order)); // 将Order对象序列化为JSON
        outboxMessage.setCreatedAt(LocalDateTime.now());
        outboxMessage.setStatus("PENDING");
        outboxMessageRepository.save(outboxMessage); // 将待发送消息写入Outbox表

        // 此时，业务数据和Outbox消息都已在同一个事务中持久化
    }

    private String serializeOrderToJson(Order order) {
        // 使用Jackson或其他库将Order对象序列化为JSON字符串
        return "{\"id\":" + order.getId() + ", \"status\":\"" + order.getStatus() + "\"}";
    }
}

// 独立的 MessageRelayerService
@Service
public class MessageRelayerService {

    private final OutboxMessageRepository outboxMessageRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public MessageRelayerService(OutboxMessageRepository outboxMessageRepository, KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxMessageRepository = outboxMessageRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 5000) // 每5秒轮询一次Outbox表
    public void relayOutboxMessages() {
        List<OutboxMessage> pendingMessages = outboxMessageRepository.findByStatus("PENDING");

        for (OutboxMessage message : pendingMessages) {
            ListenableFuture<SendResult<String, String>> future =
                    kafkaTemplate.send(message.getTopic(), message.getKey(), message.getPayload());

            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    System.out.println("Outbox消息 " + message.getId() + " 成功发送。");
                    // 标记为已发送或删除
                    message.setStatus("SENT");
                    outboxMessageRepository.save(message); // 更新状态
                }

                @Override
                public void onFailure(Throwable ex) {
                    System.err.println("Outbox消息 " + message