同消费组多消费者读取单分区Kafka Topic及并行处理方案咨询

一、单分区Topic的同组消费者消息分发规则 #

Kafka的分区是消息并行消费的最小单位，同一个消费者组内，一个分区只能被组内的一个消费者独占，不存在多个同组消费者同时消费一个分区的情况。

针对你举的例子：

• Topic A只有Partition A，组内有3个消费者时，Kafka只会把Partition A分配给其中一个消费者，另外2个消费者会处于空闲状态，不会收到任何消息。1000条消息会全部分发给这个被分配的消费者。

对应你的两个疑问：

1. 不会出现并行给3个消费者各发1条消息的情况，单分区无法被同组多消费者共享，没有这种并行分发逻辑。
2. 是的，仅由组内被分配到该分区的一个消费者获取所有消息。

二、Node.js + kafkajs实现单分区消息并行处理的最佳架构 #

因为单分区本身无法被同组多消费者并行消费，要实现4个消费者级别的并行处理，推荐两种可行方案：

方案1：单消费者进程内多Worker线程并行处理 #

利用Node.js的worker_threads模块，在单个Kafka消费者进程内，把收到的消息分发到4个Worker线程并行处理，核心逻辑：

• 启动一个kafkajs消费者，订阅目标单分区Topic
• 消费者收到消息后，将消息轮询分配给预先创建的4个Worker线程
• Worker线程处理完成后，通知主进程，主进程按顺序提交Kafka offset（避免消息重复或丢失）

示例代码

const { Kafka } = require('kafkajs')
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads')
const kafka = new Kafka({ brokers: ['localhost:9092'] })

const WORKER_COUNT = 4
const workers = []
const pendingOffsets = new Map() // 维护待确认的消息offset，保证提交顺序

// 初始化Worker线程池
if (isMainThread) {
  for (let i = 0; i < WORKER_COUNT; i++) {
    const worker = new Worker(__filename, { workerData: { workerId: i } })
    worker.on('message', ({ offset, status }) => {
      if (status === 'success') {
        pendingOffsets.delete(offset)
        // 提交已确认的最早offset，保证消费顺序
        const sortedOffsets = Array.from(pendingOffsets.keys()).sort((a, b) => a - b)
        const commitOffset = sortedOffsets.length ? sortedOffsets[0] : offset
        consumer.commitOffsets([{ 
          topic: 'topic-A', 
          partition: 0, 
          offset: (parseInt(commitOffset) + 1).toString() 
        }])
      }
    })
    workers.push(worker)
  }

  // 启动Kafka消费者
  const consumer = kafka.consumer({ groupId: 'group-A' })
  const run = async () => {
    await consumer.connect()
    await consumer.subscribe({ topic: 'topic-A', fromBeginning: true })
    await consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        const offset = message.offset
        pendingOffsets.set(offset, message)
        // 轮询分配给Worker线程
        const workerIndex = parseInt(offset) % WORKER_COUNT
        workers[workerIndex].postMessage({ payload: message.value.toString(), offset })
      },
    })
  }
  run().catch(console.error)
} else {
  // Worker线程业务处理逻辑
  parentPort.on('message', async ({ payload, offset }) => {
    try {
      console.log(`Worker ${workerData.workerId} processing: ${payload}`)
      // 替换为你的实际业务处理：比如数据库写入、API调用等
      await new Promise(resolve => setTimeout(resolve, 100)) // 模拟耗时操作
      parentPort.postMessage({ offset, status: 'success' })
    } catch (err) {
      console.error(`Worker ${workerData.workerId} failed: ${err}`)
      // 处理失败逻辑：重试、死信队列等
      parentPort.postMessage({ offset, status: 'failed' })
    }
  })
}

方案2：多消费者进程+中间多分区Topic（扩展方案） #

如果想利用Kafka原生的分区并行能力，可以新增一个多分区的中间Topic，流程：

1. 启动一个"转发消费者"，订阅原单分区Topic，把消息转发到有4个分区的中间Topic
2. 启动4个同组消费者，订阅中间Topic，每个消费者会被分配到一个分区，实现原生并行消费
3. 优点：无需手动维护线程池，Kafka自动处理负载均衡和故障转移，可靠性更高

转发消费者代码

const { Kafka } = require('kafkajs')
const kafka = new Kafka({ brokers: ['localhost:9092'] })

const consumer = kafka.consumer({ groupId: 'forward-group' })
const producer = kafka.producer()

const run = async () => {
  await producer.connect()
  await consumer.connect()
  await consumer.subscribe({ topic: 'topic-A', fromBeginning: true })
  
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      // 转发到4分区的中间Topic，按offset取模分配分区
      await producer.send({
        topic: 'topic-A-process',
        messages: [{ 
          key: message.key, 
          value: message.value,
          partition: parseInt(message.offset) % 4
        }],
      })
      // 提交原Topic的offset，保证转发可靠性
      await consumer.commitOffsets([{ topic, partition, offset: (parseInt(message.offset) + 1).toString() }])
    },
  })
}
run().catch(console.error)

处理消费者代码（启动4个进程运行此代码）

const { Kafka } = require('kafkajs')
const kafka = new Kafka({ brokers: ['localhost:9092'] })

const consumer = kafka.consumer({ groupId: 'process-group' })

const run = async () => {
  await consumer.connect()
  await consumer.subscribe({ topic: 'topic-A-process', fromBeginning: true })
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log(`Consumer on partition ${partition} processing: ${message.value.toString()}`)
      // 替换为你的实际业务处理逻辑
      await new Promise(resolve => setTimeout(resolve, 100))
    },
  })
}
run().catch(console.error)

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