您的位置:

Kafka groupid详解

一、groupid的定义

在使用Kafka的时候,我们经常会看到group.id这个配置项,它是一个字符串类型的配置项。具体来说,每个消费者都有一个group id,一般情况下我们可以将同样处理某个数据源的消费者放置在一组中,使用group id进行标识。

举个例子,如果你有一个在多个地方运行的日志处理程序,每个程序都会处理某个topic的消息,那么你可以用相同的group id来标识这个处理组,以确保传递给组中的每个处理程序的消息是唯一的。

二、groupid的作用

Kafka通过group id分配消费者之间的消息,确保一个组内的消费者不会接收到相同的消息。当同一个group id下的多个消费者订阅了同一个topic时,每个消息将只能被一个消费者消费。

在多个消费者共同消费一个topic的场景下,可以通过groupid来做load balance,即通过groupid的设置,部署多个消费者实例来对消息进行消费。

三、groupid的注意事项

1、group id需要唯一

在同一个Kafka集群中,group id需要唯一,如果两个group使用了相同的groupid,它们就会消费相同的消息,造成消息的重复消费。

2、重新启动后,groupid也需要唯一

如果在同一个group中,消费者重启或新加入消费者组,那么每次加入新消费者之前,需要确保添加的消费者的group id在之前没有被使用过。

3、group id的更改会导致消费者重新从头开始消费

Kafka集群会为group id下的每个消费者保存消费的偏移量,如果group id被更改,消费者将会从头开始消费。

四、实例代码

// 配置项
properties.put("group.id", "test-group");

// 创建消费者
KafkaConsumer consumer = new KafkaConsumer<>(properties);

// 订阅topic
consumer.subscribe(Arrays.asList("test-topic"));

// 消费消息
while (true) {
    ConsumerRecords
    records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord
     record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

    
   
  

五、小结

Kafka是一个分布式的消息队列,通过group id来保证消费者组内的消息处理具有唯一性,可以做到消息的负载均衡和处理组内消息的互斥性。在使用时需要注意group id的唯一性以及更改group id的影响等问题。