本文旨在解决 Spring EmbeddedKafka 测试场景下，生产者如何等待消费者确认消息的问题。由于 Kafka 生产者和消费者是独立的，`acks` 仅保证 Broker 接收并持久化消息，与消费者无关。因此，需要自定义逻辑实现生产者等待消费者确认的功能。本文将介绍实现此功能的思路和方法。
在 Spring EmbeddedKafka 环境下，确保生产者发送的消息被消费者正确处理并确认是一个常见的需求，尤其是在集成测试中。然而，Kafka 的设计本身是生产者和消费者解耦的，生产者端的 acks 配置仅仅控制 Broker 端的确认机制，并不能直接实现生产者等待消费者确认的功能。因此，我们需要引入额外的机制来实现这一目标。
核心思路：引入中间状态同步机制
由于生产者和消费者是独立的，我们需要一种方式让消费者在处理完消息后通知生产者。常见的做法是引入一个共享的状态存储，例如：
- 共享的 ConcurrentHashMap: 适用于单 JVM 测试环境，简单高效。
- Redis 或其他外部存储: 适用于分布式测试环境，更具扩展性。
实现步骤：
-
生产者端：
- 发送消息前，生成一个唯一的 ID (例如 UUID)。
- 将该 ID 作为消息头或者消息体的一部分发送给消费者。
- 将该 ID 存储到一个等待确认的 Map 中，并设置一个超时时间。
- 定期检查该 ID 是否从 Map 中移除，如果超时未移除，则认为消息处理失败。
-
消费者端：
- 接收到消息后，提取消息中的 ID。
- 处理完消息后，将该 ID 从共享的状态存储中移除。
- 调用 Acknowledgement.acknowledge() 确认消息已被消费。
示例代码 (使用 ConcurrentHashMap):
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 kafkaTemplate;
private final ConcurrentHashMap processedMessages = new ConcurrentHashMap<>();
public MessageHandler(KafkaTemplate 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 占用过高
}
}
} 
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 占用过高
}
}
}配置 KafkaListener:
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);
}
}注意事项：
- 超时机制： 必须设置合理的超时时间，避免无限期等待。
- 异常处理： 在消费者端，需要妥善处理消息处理失败的情况，例如重新入队或者记录错误日志。
- 消息 ID 的唯一性： 确保消息 ID 在整个系统中是唯一的。
- 并发问题： 如果多个生产者同时发送消息，需要考虑并发问题，例如使用线程安全的 Map 或者分布式锁。
总结：
虽然 Kafka 本身不提供生产者等待消费者确认的机制，但我们可以通过引入中间状态同步机制来实现这一功能。上述示例代码提供了一种基于 ConcurrentHashMap 的实现方案，可以根据实际需求选择更合适的方案，例如使用 Redis 或者其他外部存储。 关键在于生产者和消费者之间建立一个共享的状态，用于同步消息的处理状态。 通过这种方式，可以有效地提高集成测试的可靠性，确保消息被正确处理。
