从数据库向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, T> kafkaTemplate;

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

    /**
     * 以同步阻塞方式发送消息，确保严格顺序和不丢失。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表，每个元素包含一个ID和消息体
     * @return 成功发送到Kafka并得到确认的消息ID列表
     */
    public List<String> sendMessages(String topicName, List<MessageWrapper<T>> data) {
        List<String> successIds = new ArrayList<>();
        for (MessageWrapper<T> messageWrapper : data) {
            // 发送消息，并阻塞等待结果
            ListenableFuture<SendResult<String, T>> listenableFuture = 
                kafkaTemplate.send(topicName, messageWrapper.getKey(), messageWrapper.getValue());
            try {
                listenableFuture.get(3, TimeUnit.SECONDS); // 设置超时时间
                successIds.add(messageWrapper.getId()); // 成功发送，记录ID
            } catch (Exception e) {
                log.warn("消息发送失败，ID: {}，原因: {}", messageWrapper.getId(), e.getMessage());
                // 发生错误时，停止后续消息发送，确保顺序性
                break; 
            }
        }
        return successIds;
    }

    // 假设有一个包装类来获取ID和Key
    public static class MessageWrapper<T> {
        private String id;
        private String key; // 用于Kafka消息的key，通常是业务ID
        private T value;

        public MessageWrapper(String id, String key, T value) {
            this.id = id;
            this.key = key;
            this.value = value;
        }

        public String getId() { return id; }
        public String getKey() { return key; }
        public T getValue() { return value; }
    }
}

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, T> kafkaTemplate;

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

    /**
     * 以异步回调方式发送消息，提升性能，但可能牺牲严格的全局顺序。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表，每个元素包含一个ID和消息体
     * @return 成功发送到Kafka并得到确认的消息ID列表
     */
    public List<String> sendMessages(String topicName, List<SynchronousKafkaSender.MessageWrapper<T>> data) {
        // 使用线程安全的列表，因为回调可能在不同线程中执行
        List<String> successIds = Collections.synchronizedList(new ArrayList<>()); 

        data.forEach(messageWrapper ->
            kafkaTemplate.send(topicName, messageWrapper.getKey(), messageWrapper.getValue())
                    .addCallback(new ListenableFutureCallback<>() {
                        @Override
                        public void onSuccess(SendResult<String, T> result) {
                            successIds.add(messageWrapper.getId());
                        }

                        @Override
                        public void onFailure(Throwable exception) {
                            log.warn("消息异步发送失败，ID: {}，原因: {}", messageWrapper.getId(), exception.getMessage());
                            // 可以在这里添加重试逻辑或错误处理
                        }
                    }));

        // 强制刷新所有待发送的消息
        kafkaTemplate.flush(); 
        return successIds;
    }
}