优化数据库到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;

public class SynchronousKafkaSender {

    private final KafkaTemplate<String, Object> kafkaTemplate; // 假设已注入

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

    /**
     * 以同步阻塞方式发送消息，保证顺序和不丢失。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表
     * @return 成功发送并确认的消息ID列表
     */
    public List<String> sendMessagesSynchronously(String topicName, List<MyDataObject> data) {
        List<String> successIds = new ArrayList<>();
        for (MyDataObject value : data) {
            // 发送消息并获取ListenableFuture
            ListenableFuture<SendResult<String, Object>> listenableFuture = 
                kafkaTemplate.send(topicName, value.getSiebelId(), value);
            try {
                // 阻塞等待发送结果，设置超时时间
                listenableFuture.get(3, TimeUnit.SECONDS); 
                successIds.add(value.getId());
            } catch (Exception e) {
                // 如果发送失败或超时，记录警告并中断当前批次发送
                System.err.println("消息发送失败或超时，ID: " + value.getId() + ", 错误: " + e.getMessage());
                // 这里可以添加更详细的错误处理逻辑，如将失败消息记录到死信队列
                break; // 中断，未发送的消息将在下一次调度中处理
            }
        }
        return successIds;
    }
}

// 假设MyDataObject是一个包含getId()和getSiebelId()方法的自定义类
class MyDataObject {
    private String id;
    private String siebelId;
    private Object payload; // 实际数据

    public MyDataObject(String id, String siebelId, Object payload) {
        this.id = id;
        this.siebelId = siebelId;
        this.payload = payload;
    }

    public String getId() { return id; }
    public String getSiebelId() { return siebelId; }
    public Object getPayload() { return payload; }
}

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 java.util.concurrent.CopyOnWriteArrayList; // 线程安全的列表

public class AsynchronousKafkaSender {

    private final KafkaTemplate<String, Object> kafkaTemplate; // 假设已注入

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

    /**
     * 以异步回调方式发送消息，优先保证性能和不丢失，但可能牺牲局部故障时的严格顺序。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表
     * @return 成功发送并确认的消息ID列表
     */
    public List<String> sendMessagesAsynchronously(String topicName, List<MyDataObject> data) {
        // 使用线程安全的列表来收集成功发送的消息ID
        List<String> successIds = Collections.synchronizedList(new ArrayList<>()); 

        data.forEach(value -> {
            kafkaTemplate.send(topicName, value.getSiebelId(), value)
                    .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onSuccess(SendResult<String, Object> result) {
                            successIds.add(value.getId());
                            // 可以在这里记录成功日志
                        }

                        @Override
                        public void onFailure(Throwable exception) {
                            System.err.println("消息发送失败，ID: " + value.getId() + ", 错误: " + exception.getMessage());
                            // 可以在这里实现更复杂的失败处理，如重试、记录到死信队列
                        }
                    });
        });

        // 刷新KafkaTemplate，确保所有缓冲中的消息都被发送出去
        kafkaTemplate.flush(); 

        return successIds;
    }
}