node.js - 如何在node-rdkafka中一一阅读消息

我正在使用node-rdkafka（https://github.com/Blizzard/node-rdkafka）来消费消息，基本设置可以正常工作，但是每次我将某些内容推送到队列时，它都会触发该函数，而与先前方法的完成无关。

我希望在完成上一个功能时触发下一个数据单元。

这是我的实现

const Kafka = require('node-rdkafka');
const topic = 'create_user_channel';
const consumer = new Kafka.KafkaConsumer({
    'group.id':'consumer',
    'metadata.broker.list': '*******',
    'sasl.mechanisms': 'PLAIN',
    'sasl.username': '********',
    'sasl.password': '********',
    'security.protocol': 'SASL_SSL',
    'enable.auto.commit':false
}, {});

// Connect the consumer.
consumer.connect({timeout: "1000ms"}, (err) => {
    if (err) {
        console.log(`Error connecting to Kafka broker: ${err}`);
        process.exit(-1);
    }

});
let is_pause = false;
consumer.on('ready', (arg)=>{
    console.log('consumer ready.' + JSON.stringify(arg));
    console.log('Consumer is ready');
    consumer.subscribe([topic]);
    setInterval(function() {
        console.log('consumer has consume on :'+timeMs());
        consumer.consume();
      }, 1000);
});

consumer.on('data',async (data)=>{
    console.log('consumer is consuming data');
    if(!is_pause) {
        is_pause = true;
        if(data && typeof data !== 'undefined') {
            try {
                console.log('consumer received the data');
                consumer.pause([topic]);
                console.log('consumer has pause the consuming');
                await processMessage(data);
                console.log('consumer is resumed');
                consumer.resume([topic]);
                is_pause = false;
            } catch(error) {
                console.log('data consuming error');
                console.log(error);
            }
        } else {
            is_pause = false;
        }
    }
});

最佳答案

您正在调用consume()（不带任何参数），它会尽快返回消息。

如果要控制消耗速度，则可以使用其他方法consume(size)，该方法返回size Kafka记录。例如，consume(1)将返回下一个Kafka记录。

请参见node-rdkafka Consumer docs。

关于node.js - 如何在node-rdkafka中一一阅读消息，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/59928390/