查看kafka消息消费堆积情况

查看主题命令

• 展示topic列表

./kafka-topics.sh --list --zookeeper zookeeper_ip:2181

描述topic

./kafka-topics.sh --describe --zookeeper zookeeper_ip:2181 --topic topic_name

查看topic某分区偏移量最大（小）值

./kafka-run-class.sh kafka.tools.GetOffsetShell --topic topic_name --time -1 --broker-list broker_ip:9092 --partitions 0

增加topic分区数

./kafka-topics.sh --zookeeper zookeeper_ip:2181 --alter --topic test --partitions 10

删除topic：慎用，只会删除zookeeper中的元数据，消息文件须手动删除

• 方法一：

 ./kafka-topics.sh --delete --zookeeper zookeeper_ip:2181 --topic topic_name

• 方法二：
• 待验证

./kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper zookeeper_ip:2181 --topic topic_name

查看topic消费进度，必须参数为–group， 不指定–topic，默认为所有topic,

./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group group_name

列出所有主题中的所有用户组：

./kafka-consumer-groups.sh --bootstrap-server broker_ip:9092 --list

要使用ConsumerOffsetChecker查看上一个示例中消费者组的偏移量

• 按如下所示“describe”消费者组：

./kafka-consumer-groups.sh --bootstrap-server 10.1.3.84:9098 --describe --group group_name

• -members: 此选项提供使用者组中所有活动成员的列表。

./kafka-consumer-groups.sh --bootstrap-server broker_ip:9092 --describe --group group_name --members

查看kafka消息消费情况

• 消息堆积是消费滞后(Lag)的一种表现形式，消息中间件服务端中所留存的消息与消费掉的消息之间的差值即为消息堆积量，也称之为消费滞后(Lag)量。

• 对于Kafka而言，消息被发送至Topic中，而Topic又分成了多个分区(Partition)，每一个Partition都有一个预写式的日志文件，虽然Partition可以继续细分为若干个段文件(Segment)，但是对于上层应用来说可以将Partition看成最小的存储单元(一个由多个Segment文件拼接的“巨型文件”)。

• 每个Partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到Partition中。

• 我们来看下图，其就是Partition的一个真实写照：
  

• 上图中有四个概念：

• LogStartOffset：表示一个Partition的起始位移，初始为0，虽然消息的增加以及日志清除策略的影响，这个值会阶段性的增大。

• ConsumerOffset：消费位移，表示Partition的某个消费者消费到的位移位置。

• HighWatermark：简称HW，代表消费端所能“观察”到的Partition的最高日志位移，HW大于等于ConsumerOffset的值。

• LogEndOffset：简称LEO, 代表Partition的最高日志位移，其值对消费者不可见。

• 比如在ISR（In-Sync-Replicas）副本数等于3的情况下（如下图所示），消息发送到Leader A之后会更新LEO的值，Follower B和Follower C也会实时拉取Leader A中的消息来更新自己，HW就表示A、B、C三者同时达到的日志位移，也就是A、B、C三者中LEO最小的那个值。由于B、C拉取A消息之间延时问题，所以HW必然不会一直与Leader的LEO相等，即LEO>=HW。
  

• 要计算Kafka中某个消费者的滞后量很简单，首先看看其消费了几个Topic，然后针对每个Topic来计算其中每个Partition的Lag，每个Partition的Lag计算就显得非常的简单了，参考下图：
  

• 由图可知消费Lag=HW - ConsumerOffset。Kafka中自带的kafka-consumer_groups.sh脚本中就有Lag的信息，示例如下：

[root@node2 kafka_2.12-1.0.0]# bin/kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group CONSUMER_GROUP_ID