Kafkajs单文件多消费者无法并行消费问题及实现需求

首先，你的代码存在几个关键错误导致无法正常并行消费，同时Kafkajs默认采用串行处理消息，需要显式配置并行参数。以下是修正后的实现方案：

关键问题分析 #

1. 代码递归错误：funConsumer1.run实际应为consumer1.run，递归调用会导致逻辑混乱
2. 拼写错误：fromBegining应为fromBeginning
3. 手动提交偏移量缺失partition参数，会导致提交失败
4. 默认concurrency=1，单个消费者串行处理消息
5. await topicGroup1.forEach无意义，forEach是同步方法无需await

修正后的完整代码 #

const { Kafka, logLevel } = require("kafkajs");
const kafka = new Kafka({
  logLevel: logLevel.INFO,
  clientId: "kafka9845",
  brokers: ["90.45.78.123"],
  connectionTimeout: 30000,
  sessionTimeout: 30000,
  requestTimeout: 30000,
  heartbeatInterval: 10000,
  retry: {
    initialRetryTime: 5000,
    retries: 200
  }
});

const topicGroup1 = ["topic1", "topic2"];
const topicGroup2 = ["topic3", "topic4", "topic5"];

// 消费者1：处理topicGroup1，配置并行消费
const consumer1 = kafka.consumer({ groupId: 'consumer1', fromBeginning: true });
const funConsumer1 = async (consumer) => {
  // 订阅主题（forEach无需await）
  topicGroup1.forEach(topic => consumer.subscribe({ topic }));
  
  await consumer.run({
    autoCommit: false,
    concurrency: 10, // 核心：设置并行处理的消息数量，根据业务调整
    eachMessage: async (task) => {
      try {
        console.log(`Consumer1 processing ${task.topic} [partition: ${task.partition}] offset: ${task.message.offset}`);
        // --- 这里替换为你的实际业务逻辑 ---
        // await yourBusinessLogic(task.message.value.toString());
        
        // 手动提交偏移量：必须指定partition
        await consumer.commitOffsets([{
          topic: task.topic,
          partition: task.partition,
          offset: (Number(task.message.offset) + 1).toString()
        }]);
      } catch (error) {
        console.error(`Consumer1 failed to process message:`, error);
        // 可根据业务需求选择重试或跳过
      }
    }
  });
};

// 消费者2：处理topicGroup2，配置并行消费
const consumer2 = kafka.consumer({ groupId: 'consumer2', fromBeginning: true });
const funConsumer2 = async (consumer) => {
  topicGroup2.forEach(topic => consumer.subscribe({ topic }));
  
  await consumer.run({
    autoCommit: false,
    concurrency: 10,
    eachMessage: async (task) => {
      try {
        console.log(`Consumer2 processing ${task.topic} [partition: ${task.partition}] offset: ${task.message.offset}`);
        // --- 这里替换为你的实际业务逻辑 ---
        // await yourBusinessLogic(task.message.value.toString());
        
        await consumer.commitOffsets([{
          topic: task.topic,
          partition: task.partition,
          offset: (Number(task.message.offset) + 1).toString()
        }]);
      } catch (error) {
        console.error(`Consumer2 failed to process message:`, error);
      }
    }
  });
};

// 消费者崩溃重连逻辑
consumer1.on('consumer.crash', async () => {
  try {
    await consumer1.disconnect();
  } catch (error) {
    console.error('Consumer1 disconnect failed:', error);
  } finally {
    setTimeout(async () => {
      try {
        await consumer1.connect();
        await funConsumer1(consumer1);
      } catch (error) {
        console.error('Consumer1 reconnect failed:', error);
      }
    }, 5000);
  }
});

consumer2.on('consumer.crash', async () => {
  try {
    await consumer2.disconnect();
  } catch (error) {
    console.error('Consumer2 disconnect failed:', error);
  } finally {
    setTimeout(async () => {
      try {
        await consumer2.connect();
        await funConsumer2(consumer2);
      } catch (error) {
        console.error('Consumer2 reconnect failed:', error);
      }
    }, 5000);
  }
});

// 启动消费者
const funConnect = async () => {
  try {
    await Promise.all([
      consumer1.connect(),
      consumer2.connect()
    ]);
    // 同时启动两个消费者的消息处理
    await Promise.all([
      funConsumer1(consumer1),
      funConsumer2(consumer2)
    ]);
  } catch (error) {
    console.error('Failed to start consumers:', error);
    process.exit(1);
  }
};

funConnect();

// 优雅关闭
process.on('SIGINT', async () => {
  console.log('Shutting down consumers...');
  await Promise.all([
    consumer1.disconnect(),
    consumer2.disconnect()
  ]);
  process.exit(0);
});

核心优化点 #

1. 启用并行处理：通过concurrency参数指定单个消费者同时处理的消息数，根据业务耗时和服务器CPU/内存资源调整（建议10-50）
2. 修复基础错误：修正递归调用、拼写错误、偏移量提交缺失的partition参数
3. 异步启动优化：使用Promise.all同时启动两个消费者，确保真正并行运行
4. 错误处理增强：添加业务逻辑的异常捕获，避免单个消息处理失败导致消费者崩溃
5. 优雅关闭：使用Promise.all等待两个消费者断开连接后再退出进程

针对百万级消息的额外建议 #

• 分区优化：确保每个Topic的分区数足够（建议分区数 ≥ 消费者实例数 × 单实例concurrency），Kafka的并行消费能力由分区数决定
• 批量处理：对于高吞吐量场景，可使用eachBatch替代eachMessage，批量处理消息并一次性提交偏移量，减少IO开销
• Kubernetes水平扩展：部署多个Pod实例，每个Pod运行一套消费者逻辑，利用Kafka的消费者组机制自动分配分区，实现更高水平的并行消费

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