Apache Kafka:消费者组的例子
消费者组是Kafka主题中的多线程或多机消费。
消费者组
消费者可以通过使用同一个“group.id”来加入一个组。
组的最大并行度是组中消费者的数量←分区的数量。
Kafka将一个主题的分区分配给一个组中的消费者,这样每个分区只被该组中的一个消费者使用。
Kafka保证一条信息只能被群体中的一个消费者阅读。
消费者可以按照消息在日志中的存储顺序来查看消息。
消费者的重新平衡(re-balance)
添加更多的进程/线程将导致Kafka重新平衡(re-balance)。如果任何消费者或经纪人未能向ZooKeeper发送心跳,则可以通过Kafka集群对其进行重新配置。在这个重新平衡(re-balance)过程中,Kafka将把可用的分区分配给可用的线程,可能会把一个分区移动到另一个进程。
import java.util.Properties; import java.util.Arrays; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.ConsumerRecord; public class ConsumerGroup { public static void main(String[] args) throws Exception { if(args.length < 2){ System.out.println("Usage: consumer <topic> <groupname>"); return; } String topic = args[0].toString(); String group = args[1].toString(); Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", group); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("value.deserializer", "org.apache.kafka.common.serializa-tion.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); consumer.subscribe(Arrays.asList(topic)); System.out.println("Subscribed to topic " + topic); int i = 0; while (true) { ConsumerRecords<String, String> records = con-sumer.poll(100); for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value()); } } }
编译
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java
运行
>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":. ConsumerGroup <topic-name> my-group >>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":. ConsumerGroup <topic-name> my-group
在这里,我们创建了一个样本组名为my-group,有两个消费者。同样,你可以创建自己的组和定义组中的消费者数量。
输入
打开生产者命令行界面并发送一些消息,如:
Test consumer group 01 Test consumer group 02
第一个进程的输出
Subscribed to topic Hello-kafka offset = 3, key = null, value = Test consumer group 01
第二个进程的输出
Subscribed to topic Hello-kafka offset = 3, key = null, value = Test consumer group 02
现在,希望您已经通过使用Java客户端演示理解了简单消费者和消费者组。现在您已经知道了如何使用Java客户端发送和接收消息。让我们在下一章继续Kafka与大数据技术的集成。