如何通過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 
}