node.js - 使用 kafka-node 向消费者组手动提交消息

Question

我想在我的消费者组中的所有任务完成后手动提交消息(例如将消息推送到数据库)。如何禁用自动提交并手动提交消息。

Answer 1

将自动提交设置为 false

const kafka = require('kafka-node');

const config = require('../config');

const client = new kafka.KafkaClient({
  kafkaHost: config.kafkaHost
});

const consumer = new kafka.Consumer(client, [
  {
    topic: config.kafkaTopic
  }
], {
  autoCommit: false
});

然后手动提交-

consumer.on('message', (message) => {
  console.log('message', message);
  // feed data into db 
  consumer.commit((error, data) => {
    if (error) {
      console.error(error);
    } else {
      console.log('Commit success: ', data);
    }
  });
});