Kafka节点多个消费者

Question

有一个基本示例，它对1个消费者起作用。它接收消息。但是添加一个额外的消费者将被忽略。

let kafka = require('kafka-node');
let client = new kafka.Client();

let producer = new kafka.Producer(client);

let consumer1 =  new kafka.Consumer(client,[ {topic: 'topic1', partition: 0}]);
let consumer2 =  new kafka.Consumer(client,[ {topic: 'topic2', partition: 0}]);


producer.on('ready', function () {

    producer.send([{topic:'topic1', messages: 'topic 1 msg' ], (err,data)=>{
        console.log(err,'1 sent');
    });
    producer.send([{topic:'topic2', messages: 'topic 1 msg'}], (err,data)=>{
        console.log(err, '2 sent');
    });

});
producer.on('error', function (err) {
    console.log('err', err);
})



consumer1.on('message',(message) =>{
    console.log(11, message);
});
consumer2.on('message',(message) =>{
    console.log(22, message);
})

consumer2的“22”事件从未引发问题。如果我使用命令行工具检查该主题，则该主题的数据存在

Answer 1

如果kafka clusture中只有一个主题分区，则不能使用多个使用者。如果要使用n个使用者，则必须至少有n个分区。这里将对其进行更详细的解释。

Answer 2

您忘记在消费者参数中添加消费者组。在您的情况下，两个消费者都属于一个消费者组（默认情况下称为Kafka节点组）。拥有一个消费者组意味着每条消息将在每个消费者组中传递一次。由于您的主题有0个分区，因此只有第一个使用者将处理该消息。

如果您想向两个消费者发送相同的消息，您必须有两个消费者组（每个组一个消费者）。您可以通过以下方式执行：

let consumer1 =  new kafka.Consumer(client,[ {topic: 'topic1', partition: 0}], {groupId: 'group1'});
let consumer2 =  new kafka.Consumer(client,[ {topic: 'topic2', partition: 0}], {groupId: 'group2'});

如果您希望您的消费者（通过分区）一起处理消息，您需要增加主题中的分区数。