nodejs kafka consumer无限循环

wsewodh2  于 2021-06-06  发布在  Kafka
关注(0)|答案(1)|浏览(438)

我正在ubuntu16.04机器上运行kafkaèu2.11-2.0.0。创建了一个主题并从命令行界面生成了一些消息。

从命令行开始消费,消费很好。

但当我像下面这样开始nodejs消费时,它是无限迭代的。我的客户代码中有什么遗漏吗?

var kafka = require('kafka-node'),
Consumer = kafka.Consumer,
client = new kafka.Client(),
consumer = new Consumer(
    client,
    [
        {topic: 'mytopic', partition: 0}
    ],
    {
        autoCommit: true
    }
);
consumer.on('message', function (message) {
        console.log(message);
});
consumer.on('error', function (err){
        console.log(err);

})
consumer.on('offsetOutOfRange', function (err){
        console.log(err);
        process.exit();
})

这是输出。

{ topic: 'mytopic',
  value: '',
  offset: 0,
  partition: 0,
  highWaterOffset: 3,
  key: '' }
{ topic: 'mytopic',
  value: 'message2',
  offset: 1,
  partition: 0,
  highWaterOffset: 3,
  key: null }
{ topic: 'mytopic',
  value: 'message3',
  offset: 2,
  partition: 0,
  highWaterOffset: 3,
  key: null }
{ topic: 'mytopic',
  value: '',
  offset: 0,
  partition: 0,
  highWaterOffset: 3,
  key: '' }
{ topic: 'mytopic',
  value: '',
  offset: 0,
  partition: 0,
  highWaterOffset: 3,
  key: '' }
{ topic: 'mytopic',
  value: 'message2',
  offset: 1,
  partition: 0,
  highWaterOffset: 3,
  key: null }
{ topic: 'mytopic',
  value: 'message3',
  offset: 2,
  partition: 0,
  highWaterOffset: 3,
  key: null }
{ topic: 'mytopic',
  value: '',
  offset: 0,
  partition: 0,
  highWaterOffset: 3,
  key: '' }
{ topic: 'mytopic',
  value: '',
  offset: 0,
  partition: 0,
  highWaterOffset: 3,
  key: '' }
{ topic: 'mytopic',
  value: 'message2',
  offset: 1,
  partition: 0,
  highWaterOffset: 3,
  key: null }
{ topic: 'mytopic',
  value: 'message3',
  offset: 2,
  partition: 0,
  highWaterOffset: 3,
  key: null }
{ topic: 'mytopic',
  value: '',
  offset: 0,
  partition: 0,
  highWaterOffset: 3,
  key: '' }
mv1qrgav

mv1qrgav1#

最后发现Kafka的问题与新版本2.0.0有关。所以我换到了以前的版本,现在可以用了。

相关问题