Kafkajs在Node.js中消费消息过慢，如何实现并行消费？

背景与问题 #

• 基于Node.js开发Kafka消费者，涉及100个单分区Topic
• 依赖版本：kafkajs":"^2.0.0、kafka-node":"^5.0.0
• 消息生产规则：每3分钟向不同Topic推送消息
• 当前代码采用串行消费逻辑，处理效率偏低，需实现并行消费

并行消费实现方案 #

方案1：利用kafkajs内置concurrency参数快速实现并行

kafkajs的consumer.run()方法支持concurrency配置，通过启动多个worker线程并行处理不同Topic的消息（单分区Topic的消息会被分配到不同worker），无需大幅修改原有代码。

修改后的核心代码：

const funKafkaConsumer = async (kafkaConsumer) => {
  await kafkaConsumer.subscribe({ topics: topicGroups });
  await kafkaConsumer.run({
    autoCommit: false,
    concurrency: 10, // 配置并行worker数量，建议根据服务器资源调整（如10-20）
    eachMessage: async (task) => {
      console.log(task);
      await kafkaConsumer.commitOffsets([{
        topic: task.topic,
        partition: task.partition,
        offset: (Number(task.message.offset) + 1).toString()
      }]);
    }
  });
};

方案2：拆分Topic分组，启动多消费者实例

由于每个Topic是单分区，同一消费组内的多个消费者可以分配不同的Topic进行处理，最大化并行度。比如将100个Topic拆分为10组，每组启动一个消费者实例。

核心实现代码：

// 创建单个消费者实例的方法
const createConsumer = async (consumerId, assignedTopics) => {
  const consumer = kafka.consumer({ 
    groupId: 'kafkaConsumer', 
    clientId: `kafka9845-${consumerId}` // 每个实例设置唯一clientId
  });

  const runConsumer = async () => {
    await consumer.subscribe({ topics: assignedTopics });
    await consumer.run({
      autoCommit: false,
      eachMessage: async (task) => {
        console.log(`消费者${consumerId}处理消息：`, task);
        await consumer.commitOffsets([{
          topic: task.topic,
          partition: task.partition,
          offset: (Number(task.message.offset) + 1).toString()
        }]);
      }
    });
  };

  // 崩溃重连逻辑
  consumer.on('consumer.crash', async (payload) => {
    try {
      await consumer.disconnect();
    } catch (error) {
      console.error(`消费者${consumerId}断开失败：`, error);
    } finally {
      setTimeout(async () => {
        await consumer.connect();
        runConsumer().catch(console.error);
      }, 5000);
    }
  });

  await consumer.connect();
  runConsumer().catch(console.error);
  return consumer;
};

// 拆分Topic为多个分组
const splitTopicGroups = (topics, chunkSize) => {
  const chunks = [];
  for (let i = 0; i < topics.length; i += chunkSize) {
    chunks.push(topics.slice(i, i + chunkSize));
  }
  return chunks;
};

// 初始化所有消费者
const funConnect = async () => {
  const topicChunks = splitTopicGroups(topicGroups, 10); // 每组10个Topic
  await Promise.all(topicChunks.map((chunk, idx) => createConsumer(idx + 1, chunk)));
};

funConnect().catch(console.error);

process.on('SIGINT', async () => {
  console.log('收到中断信号，正在断开所有消费者...');
  // 实际项目中需保存所有consumer实例，批量执行disconnect
});

方案3：批量获取消息+异步并行处理

改用eachBatch替代eachMessage，批量拉取消息后用Promise.all并行处理，最后统一提交偏移量，适合消息处理逻辑较轻的场景。

核心代码：

const funKafkaConsumer = async (kafkaConsumer) => {
  await kafkaConsumer.subscribe({ topics: topicGroups });
  await kafkaConsumer.run({
    autoCommit: false,
    eachBatch: async ({ batch }) => {
      // 并行处理当前批次的所有消息
      await Promise.all(batch.messages.map(async (message) => {
        console.log(`处理消息：${message.value}，来自Topic：${batch.topic}`);
        // 执行你的业务处理逻辑
      }));
      // 提交当前批次最后一条消息的偏移量
      const lastOffset = batch.messages[batch.messages.length - 1].offset;
      await kafkaConsumer.commitOffsets([{
        topic: batch.topic,
        partition: batch.partition,
        offset: (Number(lastOffset) + 1).toString()
      }]);
    }
  });
};

方案选择建议 #

• 快速迭代选方案1：仅需添加concurrency参数，代码改动极小
• 最大化并行度选方案2：每个消费者独立处理部分Topic，资源利用更充分
• 批量优化选方案3：适合消息量较大、处理逻辑简单的场景

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