5 消费者组的功能

代码地址：https://github.com/luslin1711/kafka_demo/tree/master/kafka_demo_05

kafka 消费者从属于消费者群组。 一个群组里的消费者订阅的是同一主题。 每个消费者接收主题的一部分。

分区与消费者群组的数量关系：

1 分区数 > 消费者数 某些消费者可能会订阅多个分区的内容 2 分区数 = 消费者数 一个消费者订阅一个分区的内容 3 分区数 < 消费者数 一个消费者订阅一个分区的内容， 多余的消费者被闲置

测试：

topic05 分区数设置为2

Consumer.java

import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; import java.util.Collections; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class Consumer implements Runnable{ protected String groupId; protected String topic; protected Properties properties; public Consumer(String groupId, String topic) { this.groupId = groupId; this.topic = topic; this.properties = getProperties(); } protected Properties getProperties() { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", groupId); // 组id，同一组下共享offset props.put("enable.auto.commit", "false"); // 关闭自动提交 props.put("fetch.min.bytes", "512"); // 指定从服务器获取的最小字节数 props.put("fetch.max.wait.ms", "100"); // 等待时间。在服务器数据不足fetch.min.bytes时， 达到时间也会返回 props.put("auto.offset.reset", "latest"); // 在offset 失效的情况下，earliest表示从起始位置开始读取分区记录。 latest 表示从最新记录读取（在消费者启动后） return props; } public void run() { KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties, new StringDeserializer(), new StringDeserializer()); try { consumer.subscribe(Collections.singletonList(topic)); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); if (records.isEmpty()) continue; System.out.println("recordsLen: " + records.count()); for ( ConsumerRecord <String, String> record : records ) { System.out.println("threadId: " + Thread.currentThread().getName() +" ,groupId: " + groupId +", partition: " + record.partition()); } consumer.commitAsync(); } } finally { consumer.close(); } } }

子类：ConsumerWithSameGroupId

public class ConsumerWithSameGroupId extends Consumer{ public ConsumerWithSameGroupId(String groupId, String topic) { super(groupId, topic); } public static void main(String[] args) throws InterruptedException { String topic = "topic05"; ExecutorService service = Executors.newFixedThreadPool(4); for (int i = 0; i < 4; i++) { service.submit(new Consumer("group", topic)); } } } 结果： recordsLen: 8 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 recordsLen: 3 threadId: pool-1-thread-3 ,groupId: group, partition: 1 threadId: pool-1-thread-3 ,groupId: group, partition: 1 threadId: pool-1-thread-3 ,groupId: group, partition: 1 recordsLen: 3 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 threadId: pool-1-thread-4 ,groupId: group, partition: 0 ......

接收消息时会发现， 只有两个 线程有输出结果。

二、 多个消费者群组

多个消费者群组可以实现消息分发。

测试：ConsumerWithDifferentGroupId

public class ConsumerWithDifferentGroupId extends Consumer{ public ConsumerWithDifferentGroupId(String groupId, String topic) { super(groupId, topic); } public static void main(String[] args) throws InterruptedException { String topic = "topic05"; ExecutorService service = Executors.newFixedThreadPool(4); for (int i = 0; i < 4; i++) { service.submit(new Consumer("group0" + i, topic)); } } }