我正在使用 kafka-node ConsumerGroup 来消费来自主题的消息。 ConsumerGroup在消费消息时需要调用外部API,甚至可能需要一秒钟才能响应。
我希望控制消费队列中的下一条消息,直到收到 API 的响应,以便按顺序处理消息。
我应该如何控制这种行为?
这就是我们一次处理 1 条消息的方式:
var async = require('async'); //npm install async
//intialize a local worker queue with concurrency as 1 (only 1 event is processed at a time)
var q = async.queue(function(message, cb) {
processMessage(message).then(function(ep) {
cb(); //this marks the completion of the processing by the worker
});
}, 1);
// a callback function, invoked when queue is empty.
q.drain = function() {
consumerGroup.resume(); //resume listening new messages from the Kafka consumer group
};
//on receipt of message from kafka, push the message to local queue, which then will be processed by worker
function onMessage(message) {
q.push(message, function (err, result) {
if (err) { logger.error(err); return }
});
consumerGroup.pause(); //Pause kafka consumer group to not receive any more new messages
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)