Spring EmbeddedKafka 生产者等待消费者确认的实现方案

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Component
public class MessageHandler {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ConcurrentHashMap<String, Boolean> processedMessages = new ConcurrentHashMap<>();

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

    public void sendMessage(String topic, String message) throws Exception {
        String messageId = UUID.randomUUID().toString();
        processedMessages.put(messageId, false); // 添加到等待确认的 Map

        kafkaTemplate.send(topic, messageId, message).get(); // 确保消息发送成功

        // 等待消费者确认，设置超时时间
        waitForConfirmation(messageId, 5, TimeUnit.SECONDS);
    }

    public void consumeMessage(String messageId, String message, Acknowledgment acknowledgment) {
        try {
            // 处理消息...
            System.out.println("Consumed message: " + message);
            processedMessages.remove(messageId); // 移除ID，表示已确认
            acknowledgment.acknowledge();
        } catch (Exception e) {
            // 处理异常...
        }
    }

    private void waitForConfirmation(String messageId, long timeout, TimeUnit unit) throws Exception {
        long startTime = System.currentTimeMillis();
        while (processedMessages.containsKey(messageId)) {
            if (System.currentTimeMillis() - startTime > unit.toMillis(timeout)) {
                throw new Exception("Timeout waiting for message confirmation: " + messageId);
            }
            Thread.sleep(100); // 短暂休眠，避免 CPU 占用过高
        }
    }
}

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    private final MessageHandler messageHandler;

    public KafkaConsumer(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }

    @KafkaListener(topics = "your_topic", groupId = "your_group_id")
    public void listen(String messageId, String message, Acknowledgment acknowledgment) {
        messageHandler.consumeMessage(messageId, message, acknowledgment);
    }
}