如何控制ConsumerGroup处理消息的并发性

Question

我正在使用kafka-node ConsumerGroup来消费来自主题的消息。ConsumerGroup在使用消息时需要调用外部API，这可能需要一秒钟的时间来响应。我希望控制从队列中消费下一条消息，直到我从API得到响应，这样消息就会被顺序地处理。

我该如何控制这种行为？

Answer 1

这就是我们实现一次处理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
}