如何为暂未存在的对象创建监听器？同步逻辑内追踪异步Kafka请求

核心问题分析 #

你当前代码的问题在于，sendMessage里的Promise会立即执行resolve，根本没等Kafka消费者把对应requestId的数据写入myObj。而且并发场景下，多个请求的响应会互相干扰，无法精准匹配到发起请求的Promise。

正确实现方案 #

我们可以用一个Map存储每个请求对应的Promise处理器，把requestId作为键，关联对应的resolve、reject函数和超时定时器，这样就能精准匹配每个请求的响应，同时处理超时避免内存泄漏。

完整代码示例 #

// 存储待处理的请求，key为requestId，value包含resolve、reject和超时ID
const pendingRequests = new Map();

function sendMessage(requestId, data) {
  // 发送请求到Kafka主题
  kafkaProducer("requestTopic", { requestId, ...data });

  return new Promise((resolve, reject) => {
    // 设置超时逻辑，比如10秒未响应则拒绝Promise
    const timeoutId = setTimeout(() => {
      pendingRequests.delete(requestId);
      reject(new Error(`请求 ${requestId} 超时`));
    }, 10000);

    // 将当前请求的处理器存入Map
    pendingRequests.set(requestId, { resolve, reject, timeoutId });
  });
}

// Kafka消费者处理回调消息
kafkaConsumer("callbackTopic", (response) => {
  const { requestId, data } = response;
  const requestHandler = pendingRequests.get(requestId);

  if (requestHandler) {
    // 清除超时定时器
    clearTimeout(requestHandler.timeoutId);
    // 用返回的数据resolve对应的Promise
    requestHandler.resolve(data);
    // 从Map中移除已完成的请求，释放内存
    pendingRequests.delete(requestId);
  }
});

关键细节说明 #

• 精准匹配：通过requestId关联请求和响应，每个请求的Promise只会被自己对应的响应触发resolve，完全避免并发干扰。
• 超时处理：给每个请求设置超时时间，超时后自动拒绝Promise并清理Map中的记录，防止内存泄漏。
• 资源清理：响应处理完成或超时后，及时从Map中删除对应记录，避免无用数据堆积。

关于“每个请求创建独立主题/消费者”方案的分析 #

这个方案完全不可行，原因如下：

• Kafka主题是重量级资源，创建、删除主题都需要集群同步元数据，开销极大，并发请求场景下会直接拖垮集群。
• 消费者实例的创建也会占用大量资源（连接、线程等），高并发时根本无法支撑。
• Kafka的设计初衷就是通过主题+分区实现批量消息处理，单个请求对应单个主题完全违背了Kafka的设计理念，属于过度设计。

需求适配说明 #

这个方案可以完美适配你的DBaaS需求：对外暴露的API可以保持类似mongoose的同步调用风格（基于Promise的async/await），底层通过Kafka实现异步的请求响应，既替代了直接的数据库请求，又能保持现有代码结构的兼容性。

内容的提问来源于stack exchange，提问作者Caio Favero