Kafka 如何读取指定topic中的offset -------------用来验证分区是不是均衡！！！(__consumer_offsets)（注，本文尚在测试验证阶段，，，后续一俩天会追加修

Posted 2020-08-28 the_tops ----

我现在使用的是librdkafka 的C/C++ 的客户端来生产消息，用flume来辅助处理异常的数据，，，

但是在前段时间，单独使用flume测试的时候发现，flume不能对分区进行负载均衡！同一个集群中，一个broker的一个分区已经有10亿条数据，另外一台的另一个分区只有8亿条数据；

因此，我对flume参照别人的做法，增加了拦截器；

即在flume配置文件中 增加以下字段；

-----

stage_nginx.sources.tailSource.interceptors = i2
stage_nginx.sources.tailSource.interceptors.i2.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
stage_nginx.sources.tailSource.interceptors.i2.headerName=key
stage_nginx.sources.tailSource.interceptors.i2.preserveExisting=false

----特别注意，，此处的sources是你自己的sources name.

增加完后，要先进行自己测试，验证flume拦截器的负载均衡功能；

好，下来话不多少，，看测试步骤；

 

1，创建topic 相关联的分区 （因现场暂时只有2个分区，所以我这边暂时取2个分区做测试）

　　（我暂时使用的kafka版本是kafka_2.11-0.9.0.1,以下都是在kafka相关版本的bin路径下操作命令）

　　./kafka-topics.sh --create --zookeeper 192.165.1.91:12181,192.165.1.92:12181,192.165.1.64:12181 --replication-factor 1 --partitions 2 --topic test3      

　　　创建topic test3  不要分区  zookeeper 3台   分区2个  zookeeper端口号12181（我本地的broker端口号是19091，这个在kafka  conf/ server.properties里边配置）

2，查看topic的创建情况 

　　在broker的每台机器的目录下，分别查看topic的创建情况！　　

　　下边是我91机器的情况：

　　./kafka-topics.sh --describe --zookeeper 192.165.1.91:12181 --topic test3 

　　

　　　　Topic:test3 PartitionCount:2 ReplicationFactor:1 Configs:

　　　　Topic: test3 Partition: 0 Leader: 1 Replicas: 1 Isr: 1

　　　　Topic: test3 Partition: 1 Leader: 2 Replicas: 2 Isr: 2

-------------意思是  他有俩个分区，，每个分区他的备份分区都是他们自己，即没有分区，，你们可以根据你们自身的现状做不同的操作；

3，启动flume生产消息，并且查看消息是否生产成功；

　　　a, 启动flume：/home/hadoop/wgjflume/apache-flume-1.5.0-cdh5.4.9-bin/bin/flume-ng agent -n stage_nginx -c /home/hadoop/wgjflume/apache-flume-1.5.0-cdh5.4.9-bin/conf -f /home/hadoop/wgjflume/apache-flume-1.5.0-cdh5.4.9-bin/conf/flume-conf.properties -Dflume.root.logger=INFO,console      

　　　　次处生产了30条消息！！！

　　　b, 查看消息是否消费成功！

　　　./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.165.1.91:19092,192.165.1.92:19092,192.165.1.64:19092 --topic test6 --time -1

4，要查消费情况，必须的建立消费组，，下来创建消费group

　　　./kafka-console-consumer.sh --bootstrap-server 192.165.1.91:19092,192.165.1.92:19092,192.165.1.64:19092 --topic test3 --from-beginning --new-consumer

　　　　此处会消费信息！

5，（此处，关闭消费程序，或者另外开一个窗口来）查看自己创建的  group id号；

　　   ./kafka-consumer-groups.sh --bootstrap-server 192.165.1.91:19092,192.165.1.92:19092,192.165.1.64:19092 --list --new-consumer       

　　   本地我显示的是：console-consumer-54762

6，查询__consumer_offsets topic所有内容

　  　注意：运行下面命令前先要在consumer.properties中设置exclude.internal.topics=false（同时要配置好你的consumer.properties中有关zookeeper和broker相关的IP和端口信息）

　　   ./kafka-console-consumer.sh --topic __consumer_offsets --zookeeper 192.165.1.91:12181,192.165.1.92:12181,192.165.1.64:12181 --formatter "kafka.coordinator.GroupMetadataManager\\$OffsetsMessageFormatter" --consumer.config ../config/consumer.properties --from-beginning

　　此处需要注意的是  consumer.properties 的路径！！！　　

7. 计算指定consumer group在__consumer_offsets topic中分区信息

这时候就用到了第4步获取的group.id(本例中是console-consumer-54762)。Kafka会使用下面公式计算该group位移保存在__consumer_offsets的哪个分区上：

Math.abs(groupID.hashCode()) % numPartitions

所以在本例中，对应的分区=Math.abs("console-consumer-54762".hashCode()) % 50 = 22，即__consumer_offsets的分区22保存了这个consumer group的位移信息，下面让我们验证一下。

　　注意：Math.abs("console-consumer-54762".hashCode()) % 50  这个使用java输出的一个值，Math.abs是java的一个函数，可以直接将前边这个做参数，打印出他的值.

8. 获取指定consumer group的位移信息 

bin/kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 22 --broker-list 192.165.1.91:19092,192.165.1.92:19092,192.165.1.64:19092 --formatter "kafka.coordinator.GroupMetadataManager\\$OffsetsMessageFormatter"

9,生产消息，查看负载均衡情况，，，，我项目中用的flume是exec方式的，，所有使用  cat 一个文件中的内容追加到flume  exec 的文件末尾，然后运行命令8,,,查看消费位移！！！

　　结果显示，，，拦截器，，分发消息成功，，俩个分区数目基本是持衡的。

　　

 　　本文链接地址：https://i.cnblogs.com/EditPosts.aspx?postid=6339111

　　 特别鸣谢，胡夕，参考了他的博文，他的博客链接地址：http://www.cnblogs.com/huxi2b/p/6061110.html