如何在Kafkajs中实现多Consumer Group ID的连接？

问题描述 #

我们使用Kafkajs Node.js库开发了Kafka消费者应用，当前通过一个consumer group连接多个主题，代码如下：

const consumeMessages = async() => {
  const consumer = kafka.consumer({groupId : 'test-group'})
  await consumer.connect() 
  await consumer.subscribe({topic : 'test-topic', fromBeginning : true})
  await consumer.run({
    eachMessage : async({topic, partition, message}) => {
      console.log({
        value : message.value.toString(),
      })
    },
  })
}

我们希望调整为一个主题对应一个consumer group（如5个主题对应5个消费组），但尝试以下写法时无法连接到Broker并报错：

const consumer = kafka.consumer({groupId : 'test-group'}, {groupId : 'test-group1'}, {groupId : 'test-group2'})

请问如何实现多Group ID的连接？

解决方案 #

核心逻辑 #

Kafkajs的每个Consumer实例只能绑定一个Consumer Group ID，你不能通过给kafka.consumer()传多个配置对象来创建多Group的消费者，这是API使用错误。要实现一个主题对应一个消费组，必须为每个消费组创建独立的Consumer实例。

正确代码实现 #

可以封装一个通用的消费函数，批量初始化多个对应不同消费组和主题的消费者：

// 封装通用的消费初始化函数
const createConsumerForTopic = async (groupId, topic) => {
  const consumer = kafka.consumer({ groupId })
  await consumer.connect()
  await consumer.subscribe({ topic, fromBeginning: true })
  
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log(`[Group: ${groupId}, Topic: ${topic}] 收到消息:`, {
        value: message.value.toString(),
        partition
      })
    }
  })

  // 处理进程退出时的资源释放
  process.on('SIGINT', async () => {
    await consumer.disconnect()
    console.log(`消费组 ${groupId} 已断开连接`)
  })
}

// 批量启动多组消费者
const initAllConsumers = async () => {
  // 定义主题与消费组的映射关系
  const topicGroupPairs = [
    { groupId: 'test-group', topic: 'test-topic' },
    { groupId: 'test-group1', topic: 'test-topic1' },
    { groupId: 'test-group2', topic: 'test-topic2' },
    // 可继续添加更多主题-消费组对
  ]

  // 并行初始化所有消费者，提升启动效率
  await Promise.all(
    topicGroupPairs.map(pair => createConsumerForTopic(pair.groupId, pair.topic))
  )
}

// 启动所有消费者
initAllConsumers().catch(err => console.error('消费者初始化失败:', err))

关键说明 #

• 每个kafka.consumer()调用都会生成独立的消费者客户端，对应唯一的消费组ID
• 使用Promise.all并行启动多个消费者，避免串行等待带来的性能损耗
• 必须为每个消费者添加进程退出时的断开逻辑，确保Kafka连接资源正常释放

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