kafka主题offset各种需求修改方法.docVIP

kafka主题offset各种需求修改方法 　　简要：开发中，常常因为需要我们要认为修改消费者实例对kafka某个主题消费的偏移量。具体如何修改？为什么可行？其实很容易，有时候只要我们换一种方式思考，如果我自己实现kafka消费者，我该如何让我们的消费者代码如何控制对某一个主题消费，以及我们该如何实现不同消费者组可以消费同一个主题的同一条消息，一个消费组下不同消费者消费同一个主题的不同消息。如果让你实现该框架该如何实现？ 　　这里我演示实验storm的kafkaspout来进行消费，kafkaspout里面使用的低级api,所以他在zookeeper中存储数据的结构和我们使用kafka的java客户端的高级api在zookeeper中的存储结构是有所不同的。关于kafka的java客户端的高级api在zookeeper中的存储结构的构造可以看这篇文章：apache kafka系列之在zookeeper中存储结构 。 原文和作者一起讨论：/intsmaze/p/6212913.html 可接网站开发，java开发。 新浪微博：intsmaze刘洋洋哥 微信：intsmaze 　　创建一个kafka主题名为intsmazX,指定分区数为3. 　　使用kafkaspout创建该主题的消费者实例（指定元数据存放zookeeper中的路径为/kafka-offset，指定实例id为onetest），启动storm可以观察到如下信息： 复制代码 INFO storm.kafka.ZkCoordinator - Task [1/1] Refreshing partition manager connections INFO storm.kafka.DynamicBrokersReader - Read partition info from zookeeper: GlobalPartitionInformation{partitionMap={0=:6667, 1=:6667, 2=:6667}} INFO storm.kafka.KafkaUtils - Task [1/1] assigned [Partition{host=:6667, partition=0}, Partition{host=:6667, partition=1}, Partition{host=:6667, partition=2}] INFO storm.kafka.ZkCoordinator - Task [1/1] Deleted partition managers: [] INFO storm.kafka.ZkCoordinator - Task [1/1] New partition managers: [Partition{host=:6667, partition=0}, Partition{host=:6667, partition=1}, Partition{host=:6667, partition=2}] INFO storm.kafka.PartitionManager - Read partition information from: /kafka-offset/onetest/partition_0 -- null //这个地方会到zookeeper中该目录下读取，看是否存储有对该分区的消费信息 INFO storm.kafka.PartitionManager - No partition information found, using configuration to determine offset//没有分区信息，这个时候就会直接到kafka的broker中得到该分区的最大偏移量 INFO storm.kafka.PartitionManager - Last commit offset from zookeeper: 0 INFO storm.kafka.PartitionManager - Commit offset 0 is more than 9223372036854775807 behind, resetting to startOffsetTime=-2 INFO storm.kafka.PartitionManager - Starting Kafka :0 from offset 0 INFO storm.kafka.PartitionManager - Read partition information from: /kafka-offset/onetest/partition_1 -- null INFO storm.kafka.PartitionManager - No partitio