node.js - Kafka节点:慢消费者
问题描述
我正在编写一个 kafka-node 消费者,与 python 中的消费者相比,它非常慢,似乎没有收到所有消息。我试图调整所有可能的参数,但没有任何效果。
我使用这个 API:https ://github.com/SOHU-Co/kafka-node#consumer
这是我的代码:
//--------------------Create a consumer---------------------------------------------------------------------------
var kafka = require('kafka-node');
var Consumer = kafka.Consumer;
var client = new kafka.Client('192.168.2.2:2181');
var consumer = new Consumer(client,
[{ topic: 'clusterTest1', offset: 0}],
{
autoCommit: false,
fromOffset: 'latest',
outOfRangeOffset: 'latest',
fetchMinBytes: 1,
fetchMaxWaitMs: 50,
fetchMaxBytes: 52428800
}
);
offset = new kafka.Offset(client);
offset.fetch([
{ topic: 'clusterTest1', partition: 0, time: Date.now(), maxNum: 1 }
], function (err, data) {
// data
// { 't': { '0': [999] } }
});
consumer.on('message', function (message) {
console.log(message.value);
});
});
consumer.on('error', function (err) {
console.log('Error:',err);
});
consumer.on('offsetOutOfRange', function (topic) {
console.log('---------Offset out of range--------');
topic.maxNum = 2;
offset.fetch([topic], function (err, offsets) {
if (err) {
return console.error(err);
}
var max= Math.max.apply(null, offsets[topic.topic][topic.partition]);
consumer.setOffset(topic.topic, topic.partition, max);
});
});
但是,这个 python 代码工作正常:
from kafka import KafkaConsumer
consumer = KafkaConsumer(bootstrap_servers=['192.168.2.2:9092','192.168.2.3:9092','192.168.2.4:9092','192.168.2.5:9092'])
print('Erfolgreich verbunden')
print(consumer.subscribe('clusterTest1'))
print(consumer.subscription())
#print(consumer.DEFAULT_CONFIG)
for msg in consumer:
print(msg.value)
感谢您提前提供任何帮助。
解决方案
解决了。我不知道有多个分区。现在,我正在使用消费者组,我可以接收所有消息。