如何通过ConsumerGroup控制处理消息的并发性 [英] How to control the concurrency of processing messages by ConsumerGroup
本文介绍了如何通过ConsumerGroup控制处理消息的并发性的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我正在使用kafka-node ConsumerGroup来消费主题中的消息. ConsumerGroup在使用一条消息时需要调用一个外部API,这甚至可能需要一秒钟的时间才能响应. 我希望控制队列中的下一条消息的消耗,直到我从API获得响应为止,以便按顺序处理这些消息.
I am using kafka-node ConsumerGroup to consume message from a topic. The ConsumerGroup when consumes a message requires calling an external API, which might even take a second to response. I wish to control consuming next message from the queue until I get the response from the API, so that the messages are processed sequentially.
我应如何控制此行为?
How should I control this behavior?
推荐答案
这是我们一次实现处理1条消息的方式:
This is how we have implemented processing of 1 message at a time:
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
}
这篇关于如何通过ConsumerGroup控制处理消息的并发性的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文