ConsumerGroup如何控制处理消息的并发 [英] How to control the concurrency of processing messages by 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屋！