NestJS+Kafka多生产者单消费者请求响应超时问题排查

核心原因 #

Kafka消费者组（groupId）的分区分配机制导致响应分发异常：

• 当多个消费者实例使用同一个groupId订阅同一主题时，Kafka会将主题的分区均匀分配给组内的消费者，每个分区仅由组内一个实例消费。
• 你的两个生产者同时订阅响应主题且共用groupId，消费者服务返回的响应会被Kafka随机分发到组内某一个生产者实例（先启动的producer1），producer2无法收到自己请求对应的响应，最终触发超时。

解决方法 #

1. 为每个生产者分配独立的groupId #

让每个生产者作为响应主题的消费者时使用唯一的groupId，这样Kafka会将响应主题的所有分区分配给每个生产者实例，确保每个生产者都能收到自己请求的响应。
示例代码修改（NestJS Kafka配置）：

// producer1的Kafka配置
@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_CLIENT',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'producer1-client',
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'producer1-response-group', // 独立groupId
          },
        },
      },
    ]),
  ],
})
export class Producer1Module {}

// producer2的Kafka配置
@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_CLIENT',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'producer2-client',
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'producer2-response-group', // 独立groupId
          },
        },
      },
    ]),
  ],
})
export class Producer2Module {}

2. 添加请求关联标识做响应过滤（可选补充） #

发送请求时携带唯一correlationId，生产者消费响应主题时仅处理匹配自身请求标识的消息，进一步避免消息混淆：

• 发送请求时携带标识：

async sendData(data: any) {
  const correlationId = uuidv4(); // 生成唯一标识
  return this.client.send('send_data', {
    data,
    correlationId,
  });
}

• 生产者消费响应时过滤：

@SubscribeResponse('response_topic')
async handleResponse(@Payload() payload: any) {
  const correlationId = payload.correlationId;
  // 仅处理当前生产者发起的请求对应的响应
  if (this.pendingRequests.has(correlationId)) {
    // 执行响应处理逻辑
    this.pendingRequests.delete(correlationId);
  }
}

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