Kafka消息最后两条顺序异常求助（NestJS+KafkaJS场景）

核心问题定位 #

你遇到的顺序颠倒问题，本质是循环中异步发送未等待前一个请求完成，导致多个消息发送请求并发执行，即使设置了maxInFlightRequests:1，也因为NestJS Kafka客户端的异步封装逻辑，无法保证发送顺序的串行性。加sleep能恢复顺序，就是因为强制给并发请求加了串行等待的时间窗口。

具体排查与修复步骤 #

1. 强制消息串行发送 #

修改OrderService中的循环逻辑，确保每个消息发送完成后再发送下一个：

// OrderService
// 确保sendMessage是异步函数，内部await Kafka发送操作
for (const order of productionOrders) {
  await this.sendMessage(
    kafkaUtilities.buildOrderedMessage(
      'my-topic-here', Array.of(order),
      'my-key-here')
  );
}

同时修改KafkaService的sendMessage方法，确保等待Kafka发送完成：

// KafkaService
async sendMessage(message) {
  await this.kafkaClient.emit(message.topic, message.messages);
}

2. 确认KafkaJS Producer配置生效 #

检查NestJS Kafka客户端的producer配置，确保maxInFlightRequests:1正确设置（该配置限制每个连接的未完成请求数，保证同一连接的请求串行处理）：

// 在ClientsModule注册时添加producer配置
ClientsModule.register([
  {
    name: 'KAFKA_CLIENT',
    transport: Transport.KAFKA,
    options: {
      client: {
        clientId: 'your-client-id',
        brokers: ['your-broker-address'],
      },
      producer: {
        maxInFlightRequests: 1, // 必须确保该配置生效
      },
    },
  },
]);

3. 验证消息的Partition绑定 #

Kafka仅保证同一Partition内的消息顺序，需确认：

• 相同key的消息是否被路由到同一个Partition：可以在消费端打印消息的partition字段，验证同key消息的Partition一致性。
• 若手动指定Partition，确保所有需要顺序的消息都发送到同一个Partition：

  // KafkaUtilities的createMessage方法中添加指定Partition
  private createMessage(topic: string, data: Array<unknown>, key?: string): ProducerRecord {
    const content = {
      value: JSON.stringify(data),
      partition: 0, // 指定固定Partition，确保同顺序消息进入同一分区
    } as any
  
    if (key) {
      this.timestamp += 1000
      content.key = key
      content.timestamp = this.timestamp.toString()
    }
  
    return { topic, messages: [content] } // 注意：messages应该是数组，原代码可能有误！
  }
  

  （注：原代码中messages: content是错误的，KafkaJS的ProducerRecord要求messages是数组类型，这可能也是导致顺序问题的潜在原因）

4. 升级KafkaJS版本 #

你使用的KafkaJS 1.15.0是较旧的版本，存在一些并发发送的已知问题。建议升级到最新稳定版（如2.x系列），再验证问题是否消失。

5. 跳过NestJS封装，直接使用KafkaJS原生API #

若怀疑NestJS的Kafka客户端封装导致异步逻辑异常，可以绕过封装，直接用KafkaJS原生Producer发送消息：

// 注入原生Kafka Producer实例
async sendRawMessage(topic: string, data: any, key: string) {
  await this.producer.send({
    topic,
    messages: [
      {
        key,
        value: JSON.stringify(data),
        partition: 0,
      },
    ],
  });
}

6. 检查消费端配置 #

确认消费端未开启并发消费：若消费端使用多线程/多消费者实例，即使Kafka中消息顺序正确，消费时也可能乱序。需保证同一Partition的消息由同一个Consumer实例处理。

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