从数据库到Kafka的可靠消息传输：兼顾顺序与性能的策略解析

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SynchronousKafkaSender<T> {

    private static final Logger log = LoggerFactory.getLogger(SynchronousKafkaSender.class);
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public SynchronousKafkaSender(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 同步发送消息到Kafka，严格保证顺序和不丢失。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表
     * @return 成功发送并确认的消息ID列表
     */
    public List<String> sendMessagesSynchronously(String topicName, List<T> data) {
        List<String> successIds = new ArrayList<>();
        for (T value : data) {
            // 假设T对象有一个getId()方法获取唯一标识
            String messageId = ((MyDataClass) value).getId(); // 替换为实际的数据类型和ID获取方式
            String siebelId = ((MyDataClass) value).getSiebelId(); // 替换为实际的key获取方式

            ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(topicName, siebelId, value);
            try {
                // 阻塞等待发送结果，设置超时时间
                listenableFuture.get(3, TimeUnit.SECONDS);
                successIds.add(messageId);
            } catch (Exception e) {
                log.warn("消息发送失败，ID: {}，错误信息: {}。停止当前批次后续消息发送。", messageId, e.getMessage());
                // 如果当前消息发送失败，则中断当前批次的发送，等待下一次调度重试
                break;
            }
        }
        return successIds;
    }

    // 示例数据类，实际应用中替换为您的业务数据类
    private static class MyDataClass {
        private String id;
        private String siebelId;
        private String content;

        public MyDataClass(String id, String siebelId, String content) {
            this.id = id;
            this.siebelId = siebelId;
            this.content = content;
        }

        public String getId() { return id; }
        public String getSiebelId() { return siebelId; }
        public String getContent() { return content; }
    }
}

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class AsynchronousKafkaSender<T> {

    private static final Logger log = LoggerFactory.getLogger(AsynchronousKafkaSender.class);
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public AsynchronousKafkaSender(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 异步发送消息到Kafka，优化性能，但可能对严格顺序性有所妥协。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表
     * @return 成功发送并确认的消息ID列表
     */
    public List<String> sendMessagesAsynchronously(String topicName, List<T> data) {
        // 使用线程安全的列表，因为回调可能在不同线程中执行
        List<String> successIds = Collections.synchronizedList(new ArrayList<>());

        data.forEach(value -> {
            String messageId = ((MyDataClass) value).getId(); // 替换为实际的数据类型和ID获取方式
            String siebelId = ((MyDataClass) value).getSiebelId(); // 替换为实际的key获取方式

            kafkaTemplate.send(topicName, siebelId, value)
                    .addCallback(new ListenableFutureCallback<>() {
                        @Override
                        public void onSuccess(SendResult<String, Object> result) {
                            successIds.add(messageId);
                            log.debug("消息发送成功，ID: {}，偏移量: {}", messageId, result.getRecordMetadata().offset());
                        }

                        @Override
                        public void onFailure(Throwable exception) {
                            log.warn("消息发送失败，ID: {}，错误信息: {}", messageId, exception.getMessage());
                            // 这里可以添加更复杂的错误处理逻辑，例如记录到单独的失败队列
                        }
                    });
        });

        // 刷新KafkaTemplate，确保所有缓冲区的消息被发送
        // 这一步是提升性能的关键，它允许批量发送
        kafkaTemplate.flush();
        log.info("当前批次所有消息已提交发送请求并刷新KafkaTemplate。");

        // 注意：这里的successIds列表可能不会立即包含所有成功发送的消息ID，
        // 因为回调是异步执行的。在实际应用中，如果需要等待所有回调完成，
        // 需要更复杂的同步机制（如CountDownLatch）。
        // 对于本场景，successIds用于标识“已成功发送的请求”，
        // 即使回调尚未完成，flush()也确保了消息被推送到Kafka。
        // 下一次调度时，未在successIds中的消息会被重新处理。
        return successIds;
    }

    // 示例数据类，实际应用中替换为您的业务数据类
    private static class MyDataClass {
        private String id;
        private String siebelId;
        private String content;

        public MyDataClass(String id, String siebelId, String content) {
            this.id = id;
            this.siebelId = siebelId;
            this.content = content;
        }

        public String getId() { return id; }
        public String getSiebelId() { return siebelId; }
        public String getContent() { return content; }
    }
}