一、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的影响等问题。
原创文章，作者：CMBBC，如若转载，请注明出处：https://www.506064.com/n/369097.html
微信扫一扫 
支付宝扫一扫 