KAFKA-NODE：管理多个消费者组

Question

我有一个关于管理多个 CG 的问题，创建了三个消费者组，每个 CG 都有自己的 kafka 服务、组 ID 和主题。

现在我正在按预期接收消息，但是，我想知道是否可以创建下一个场景：

创建三个消费者组，但只接收来自一个的消息，暂时让其他人暂停/保持，如果他的kafka服务将下降，消费来自下一个消费者组的消息，与第三个相同。

这是我的代码示例：

function createConsumerGroup(topics){

    const ConsumerGroup = kafka.ConsumerGroup;

    //CREATE CONSUMER GROUPS FOR EVERY SERVICE
    for(let i = 0; i < config.kafka_service.length ;i++){  //3

        const options = {
            groupId: config.kafka_service[i]['groupId'],
            host: config.kafka_service[i]['zookeeperHost'],
            kafkaHost: config.kafka_service[i]['kafkaHost'],
            sessionTimeout: 15000,
            protocol: ['roundrobin'],
            fromOffset: 'latest'
        }

        //assign all services CG names and create [i] consumer groups!
        let customConsumerGroupName = config.kafka_service[i]['consumerGroupName'];

        customConsumerGroupName = new ConsumerGroup(options, topics);

        customConsumerGroupName.on('connect', (resp) => {
            console.log(`${config.kafka_service[i]['consumerGroupName']} is connected!`);
        });

        if(i > 0){
            //pause consumers exept FIRST
            customConsumerGroupName.pause();
        }


        customConsumerGroupName.on('message', (message) => {
           console.log(message);
        });

        customConsumerGroupName.on('error', (error) => {
            console.log('consumer group error: ', error);

           //HERE I NEED TO CALL SECOND CONSUMER TO STEP UP
           //MAYBE consumerGroup.resume(); ???
        });

    }
}

希望它的声音可以理解，谢谢 :)

Answer 1

所以看起来混淆是因为 Node 包的“ConsumerGroup”的名称。在 Kafka 术语中，消费者组仅由每个消费者使用的 groupId 控制。具有相同 groupId 的消费者不会收到重复消息，每个主题消息仅由单个消费者读取。如果消费者出现故障，kafka 会检测到这一点并将其分区提供给单独的消费者。

节点“ConsumerGroup”实际上只是另一个 Kafka 消费者（新的消费者组由 Kafka 管理，而不是 Kafka >0.9 的 zookeeper）。

因此，通过 Node ConsumerGroup 来利用 kafka 消费者组的方法如下：

function createConsumerGroup(topics){

const ConsumerGroup = kafka.ConsumerGroup;

//CREATE CONSUMER GROUPS FOR EVERY SERVICE
for(let i = 0; i < config.kafka_service.length ;i++){  //3

    const options = {
        groupId: 'SOME_GROUP_NAME',
        host: config.kafka_service[i]['zookeeperHost'],
        kafkaHost: config.kafka_service[i]['kafkaHost'],
        sessionTimeout: 15000,
        protocol: ['roundrobin'],
        fromOffset: 'latest'
    }

    //assign all services CG names and create [i] consumer groups!
    let customConsumerGroupName = config.kafka_service[i]['consumerGroupName'];

    customConsumerGroupName = new ConsumerGroup(options, topics);

    customConsumerGroupName.on('connect', (resp) => {
        console.log(`${config.kafka_service[i]['consumerGroupName']} is connected!`);
    });

    customConsumerGroupName.on('message', (message) => {
       console.log(message);
    });

    customConsumerGroupName.on('error', (error) => {
        console.log('consumer group error: ', error);

       //Error handling logic here, restart the consumer that failed perhaps? 
       //Depends on how you want to managed failed consumers.
    });
  }
}

Nodes ConsumerGroup 的每个实例都将是组“SOME_GROUP_NAME”的成员，并且使用相同 groupId 创建的任何其他消费者也将充当同一个 kafka 消费者组的成员，无论服务器如何等等。

Answer 2

消费者群体解决了两个核心场景：

1.缩放 您可以增加组中的消费者数量，以处理该组正在消费（向外扩展）的主题中不断增加的消息生成率

2。故障转移 通过让一组消费者阅读相同的主题，他们将自动处理一个或多个消费者宕机的情况。

因此，您不必拥有“备用”消费者组，您必须自己处理哪些是活跃的，您只需依靠 Kafka 的内置故障转移。消费者可以在多个不同的容器中运行（甚至在不同的数据中心），并且 Kafka 将自动确保将消息传递给各个消费者，无论他们在哪里或在任何给定时间有多少人在运行。