Spring EmbeddedKafka 生产者等待消费者确认的实现方法_技术教程

本文介绍了在使用 Spring EmbeddedKafka 进行集成测试时，如何实现生产者等待消费者确认消息已被处理的机制。由于 Kafka 的生产者和消费者是独立的，`acks` 参数仅保证 Broker 收到并持久化消息，并不能确保消费者成功消费。因此，我们需要自定义逻辑来实现生产者对消费者确认的等待。

在使用 Spring EmbeddedKafka 进行集成测试时，一个常见的需求是确保生产者发送的消息被消费者成功处理后再进行后续操作。然而，Kafka 的设计是生产者和消费者完全解耦的，生产者无法直接感知消费者是否已经消费了消息。即使配置了 acks=all，也只能保证消息被 Broker 成功接收并持久化，而不能保证消费者已经调用 Acknowledgement.acknowledge() 方法确认消费。

理解 Kafka 的 acks 参数

Kafka 的 acks 参数控制着生产者发送消息后，需要多少个 Broker 确认接收消息才算成功。它有以下几种取值：

acks=0: 生产者发送消息后，不会等待任何 Broker 的确认。这种模式下吞吐量最高，但可靠性最低，消息可能丢失。
acks=1: 生产者发送消息后，等待 Leader Broker 确认接收消息。这种模式下吞吐量和可靠性之间取得平衡。
acks=all 或 acks=-1: 生产者发送消息后，等待所有 ISR (In-Sync Replicas) Broker 确认接收消息。这种模式下可靠性最高，但吞吐量最低。

需要注意的是，acks 参数只与 Broker 的确认机制有关，与消费者是否消费消息无关。

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

由于 Kafka 本身不提供生产者等待消费者确认的机制，我们需要自定义逻辑来实现。以下是一些可能的方案：

使用共享状态: 可以通过共享内存、数据库、Redis 等方式，让消费者在消费消息后更新一个状态，生产者轮询检查该状态，直到状态变为已消费。
- 优点: 实现简单。
- 缺点: 需要引入额外的组件，轮询检查会消耗资源，可能存在延迟。
使用 Kafka Topic 作为确认通道: 消费者在成功消费消息后，向特定的 Kafka Topic 发送一条确认消息，生产者监听该 Topic，收到确认消息后认为消息已被消费。
- 优点: 利用 Kafka 的消息机制，可靠性较高。
- 缺点: 需要配置额外的 Topic，增加了复杂度。
使用回调函数: 生产者发送消息时，传递一个回调函数给消费者，消费者在成功消费消息后执行该回调函数。
- 优点: 实现简单，不需要额外的组件。
- 缺点: 回调函数的执行可能会阻塞消费者线程，需要谨慎处理。

示例代码 (使用 Kafka Topic 作为确认通道)

以下是一个使用 Kafka Topic 作为确认通道的示例代码：

生产者:

@Autowired
private KafkaTemplate kafkaTemplate;

private final String confirmationTopic = "confirmation-topic";

public void sendMessageAndWaitForConfirmation(String topic, String message, String correlationId) throws ExecutionException, InterruptedException, TimeoutException {
    // 1. 发送消息到目标 Topic
    kafkaTemplate.send(topic, message);

    // 2. 监听确认 Topic，等待确认消息
    MessageListenerContainer container = createConfirmationListenerContainer(correlationId);
    container.start();

    // 3. 设置超时时间，防止无限等待
    try {
        latch.await(10, TimeUnit.SECONDS); // 假设超时时间为 10 秒
    } finally {
        container.stop(); // 停止监听器
    }

    if (latch.getCount() > 0) {
        throw new TimeoutException("Timeout waiting for confirmation message.");
    }
}

private MessageListenerContainer createConfirmationListenerContainer(String correlationId) {
    ContainerProperties containerProps = new ContainerProperties(confirmationTopic);
    CountDownLatch latch = new CountDownLatch(1);
    containerProps.setMessageListener((MessageListener) record -> {
        if (record.value().equals(correlationId)) {
            latch.countDown();
        }
    });

    ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
    return container;
}

消费者:

@KafkaListener(topics = "your-topic", groupId = "your-group")
public void listen(String message, Acknowledgment acknowledgment, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) {
    // 1. 处理消息
    processMessage(message);

    // 2. 发送确认消息到确认 Topic
    String correlationId = key; // 使用消息的 Key 作为 Correlation ID
    kafkaTemplate.send("confirmation-topic", correlationId);

    // 3. 手动提交 Offset
    acknowledgment.acknowledge();
}

注意事项: