如何使用kafkajs按指定数量批量消费Kafka Topic消息

核心思路 #

要实现按指定数量分批消费，核心是手动控制消息拉取数量与偏移量提交：

• 禁用自动提交偏移量，改为手动提交，确保每批消息处理完成后再更新消费位置
• 每次调用consumer.fetch()拉取指定数量的消息
• 处理完当前批次后，提交对应偏移量，再进入下一轮拉取循环

具体实现代码 #

首先初始化Kafka消费者：

const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'batch-consumer-app',
  brokers: ['localhost:9092'] // 替换为你的Kafka Broker地址
})

const consumer = kafka.consumer({ 
  groupId: 'batch-consumer-group',
  autoCommit: false // 必须禁用自动提交，手动控制偏移量
})

然后编写分批消费逻辑：

async function batchConsume(topic, batchSize, totalMessages) {
  await consumer.connect()
  await consumer.subscribe({ topic, fromBeginning: true })

  let consumedCount = 0

  while (consumedCount < totalMessages) {
    // 拉取指定数量的消息
    const messages = await consumer.fetch({
      topic,
      maxWaitTimeInMs: 1000, // 无消息时的最长等待时间，避免无限阻塞
      maxBytes: 1024 * 1024, // 单批次消息最大字节数，可按需调整
      maxNumberOfMessages: batchSize // 每次拉取的最大消息数
    })

    if (messages.length === 0) {
      console.log('当前无新消息，2秒后重试...')
      await new Promise(resolve => setTimeout(resolve, 2000))
      continue
    }

    // 处理当前批次消息
    console.log(`处理第${Math.floor(consumedCount/batchSize)+1}批次，共${messages.length}条消息`)
    for (const message of messages) {
      // 这里替换为你的业务处理逻辑
      console.log(`处理内容：${message.value.toString()}`)
      consumedCount++
    }

    // 提交当前批次的偏移量
    const lastMsg = messages[messages.length - 1]
    await consumer.commitOffsets([{
      topic,
      partition: lastMsg.partition,
      offset: (parseInt(lastMsg.offset) + 1).toString() // 提交下一个待消费的偏移量
    }])

    console.log(`第${Math.floor(consumedCount/batchSize)}批次处理完成，累计消费${consumedCount}条`)
  }

  await consumer.disconnect()
  console.log('所有消息处理完毕，消费者已断开连接')
}

// 调用示例：消费test-topic，每批100条，总计1000条
batchConsume('test-topic', 100, 1000).catch(console.error)

关键细节说明 #

• autoCommit: false：必须禁用自动提交，否则Kafka会自动更新偏移量，导致分批逻辑失效
• consumer.fetch()：手动拉取消息，通过maxNumberOfMessages精准控制每批数量，maxWaitTimeInMs避免空等
• 偏移量提交：提交的偏移量为当前批次最后一条消息的偏移量+1，确保下一轮从正确位置开始消费
• 循环终止：通过consumedCount累计已消费数量，达到totalMessages后停止循环

注意事项 #

• 若Topic存在多分区，上述示例仅处理单分区场景，如需支持多分区，需遍历每个分区单独拉取和提交偏移量
• 可根据业务场景调整maxBytes和maxWaitTimeInMs参数，平衡拉取效率与及时性
• 消息处理出现异常时，需根据业务需求决定是否重试或跳过，避免错误提交偏移量导致消息丢失或重复消费

内容的提问来源于stack exchange，提问作者M.S.Udhaya Raj