Kafka权威指南第四章第11节:独立消费者-为什么、怎样在没有消费组的情况下使用消...

问题导读

1.为什么需要在没有消费组的情况下使用独立消费者？
2.怎样没有消费组的情况下使用独立消费者？
3.具体代码如何编写？

到目前为止，我们讨论了可自动分配分区到消费者并且组中消费者新增或删除会自动触发负载均衡的消费组。通常，这种行为正是你想要的，但在某些情况下，却想要更简单的东西。有时你知道你有一个单一的消费者总是需要从主题中的所有分区读取数据，或者从一个主题特定分区读取数据。在这种情况下没有理由需要组或负载均衡，只是订阅特定的主题或分区，偶尔使用消息和提交偏移量。

在这中场景下，你不需要订阅一个主题，取而代之的是你只需要分配给自己几个分区。这是一个样例，一个消费者如何分配给自己特定主题的几个分区，并且从中消费：

[mw_shl_code=java,false]List<PartitionInfo> partitionInfos = null;
partitionInfos = consumer.partitionsFor("topic");
if (partitionInfos != null) {
for (PartitionInfo partition : partitionInfos)
partitions.add(new TopicPartition(partition.topic(), partition.partition()));
consumer.assign(partitions);
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record: records) {
System.out.println("topic = %s, partition = %s, offset = %d, customer
= %s, country = %s\n",
record.topic(), record.partition(), record.offset(),
record.key(), record.value());
}
consumer.commitSync();
}
}
[/mw_shl_code]

值得注意的是，除了没有负载均衡以及需要手动查找分区，一切看起来都很正常。请记住，如果有人向主题添加新分区，则不会通知消费者。所以无论是处理通过定期检查consumer.partitionsFor()或者记住是否是管理员添加分区，应用程序将需要跳跃。还要注意的是消费者可以订阅的主题（成为一个消费组的一部分），或分配自己的分区，但不能同时实现。