如何在Kafka的两个API间实现15分钟消息响应超时机制？

核心思路 #

基于请求-响应的关联标识+本地/分布式超时追踪缓存实现，不需要新增Kafka Topic。

1. 给请求消息绑定唯一关联ID #

API1生产消息到第一个Topic时，在消息Headers（推荐，不侵入业务 payload）或消息体中加入全局唯一的requestId（比如UUID），用来关联后续的响应消息。

• 示例Headers结构：

  Headers: [
    Key: "request-id", Value: "a1b2c3d4-5678-90ef-ghij-klmnopqrstuv"
  ]
  Payload: { ... 业务数据 ... }
  

2. API1维护超时追踪缓存 #

API1发送消息成功后，立即将requestId和当前时间戳存入缓存：

• 单实例部署可用Guava Cache等内存缓存，集群部署必须用Redis这类分布式缓存
• 缓存过期时间直接设为15分钟
• 伪代码示例：

  String requestId = UUID.randomUUID().toString();
  // 发送消息到Topic1
  kafkaProducer.send(new ProducerRecord<>("topic1", null, requestId, payload), (metadata, exception) -> {
      if (exception == null) {
          // 存入缓存并设置15分钟过期
          cache.put(requestId, System.currentTimeMillis(), Duration.ofMinutes(15));
      }
  });
  

3. 响应消息携带关联ID #

API2消费Topic1的消息后，提取requestId，并将其带入发送到Topic2的响应消息中，确保API1能通过requestId匹配到对应的请求。

4. 消费响应时校验超时状态 #

API1消费Topic2的响应消息时，提取requestId并检查缓存：

• 缓存存在：说明15分钟内收到响应，清理缓存记录，正常处理业务
• 缓存不存在：说明响应超时，触发超时逻辑（比如记录告警、标记请求失败）

5. 利用缓存过期回调主动处理超时 #

通过缓存的过期回调机制（Redis Key过期事件、Guava Cache的removalListener），主动对过期的requestId触发超时处理，无需轮询：

• 示例Redis过期回调逻辑：

  redisClient.addListener((key, value) -> {
      if (key.startsWith("kafka-request-")) {
          String requestId = key.substring("kafka-request-".length());
          // 执行超时处理：日志记录、告警、更新业务状态等
          handleRequestTimeout(requestId);
      }
  });
  

关键注意事项 #

• 集群部署必须用分布式缓存，避免跨节点请求无法追踪
• 处理重复响应：若API1收到重复响应，此时缓存已被清理，直接忽略即可
• 消息丢失场景：Topic1消息丢失或Topic2响应丢失，都会触发缓存过期的超时逻辑，可根据业务需求添加重试机制

内容的提问来源于stack exchange，提问作者Kushal Reddy