Kafka是一种高吞吐量的分布式发布订阅消息系统，架构包含producer(生产者)、broker、consumer(消费者)，是目前最流行的消息队列之一，广泛应用在微服务拆分中。

producer负责生产消息，然后投递到Kafka broker （即Kafka server）中，consumer 负责消费消息，从broker上定时拉取消息。比如一个电商支付、发货流程的消息流程如下

producer和consumer是没有直接关系的，中间以broker作为纽带关联起来，双方的消息规范人为协定，比如 topic、消息体内容、序列化方式等等，producer、consumer 的这些基本参数在上篇《spring kafka 写生产者消费者及参数详解》已经详细介绍。

在我们实际工作中可能会出现种种问题，比如生产者发送消息没有真正成功、消费者没有消费到消息或者重复消费了，那么该如何去 broker 上查看消息，排查问题？稍微大一些的公司都有架构组或者中台组做好的 Kafka 管理后台查看topic、消费组、消息阻塞情况，但也似乎很少提供查询具体到某条消息的。本文就是以最接近真相的方式去查看kafka原始数据。以下内容主要包括：

1. kafka 服务注册信息
2. 查看 topic
3. 查看 group
4. 查看 kafka 消息内容

01. kafka 服务注册信息

kafka 的producer、broker信息、consumer 基本关系图如下

kafka 服务信息是注册在 zookeeper 上的，包括选主也是在 zookeeper 上做的。

1.1. 查看 kafka 进程

运行 ps -ef |grep kafka 后可以看到 kafka 路径和 kafka.properties

kafka.properties 是服务的配置信息，我们关心zookeeper.connect 和 log.dirs。

1. zookeeper.connect 是kafka使用的zookeeper的host
2. log.dirs 是保存 kafka log 文件的路径

1.2. zookeeper 保存了哪些信息

登陆上面的zookeeper，kafka信息在Zookeeper 里的存储结构是这样的，主要保存了broker信息、topic信息。

• broker 和 topic

有三个broker id分别为 35、36、37，35信息如下

• controller

controller 保存了 kafka leader 信息

02. 查看topic

2.1 脚本

在执行ps -ef |grep kafka 后可以看到 ..KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/bin/,这里面存放了 Kafka 提供的脚本，我将使用这些脚本查看topic、group、消息内容。

2.2 list topic

topic 是注册在 Zookeeper 上的，你可以去 Zookeeper 上查看，也可以使用 kafka bin脚本查看。使用 kafka bin 脚本查看方式，如下：

sh kafka-topics.sh --zookeeper Zookeeper_host:port --list

可以列举出所有的topic

2.3 查看特定的topic

sh kafka-topics.sh --zookeeper Zookeeper_host:port --topic topic_name --describe

03. 查看group

3.1 list group

kafka 是通过 group 来分组消费的，注册kafka也要指定一个 groupId，通过 --list 可以查看自己的 groupId 是否注册到了 kafka，以及 查看 groupId列表。

有时候会出现在代码中向 kafka 注册了 consumer group，但一直消费不到消息，这时可以看一下你的group是否注册成功了。

sh kafka-consumer-groups.sh --bootstrap-server kafka_server:port --list

3.2 查看 group - topic 的消费进度

如果consumer迟迟消费不到消息或者消费延迟，可以看一下你的group下某个topic是否有消费阻塞。

sh kafka-consumer-groups.sh --bootstrap-server server:port --group groupId --describebr

PARTITION：分区

CURRENT-OFFSET：当前消费到的 offset

LOG-END-OFFSET：最新消息的offset

LAG：滞后多少个offset，即阻塞多少个

04. 查看 Kafka 消息内容

kafka.properties有一项配置是 log.dirs，这就是保存消息的地方，Kafka 以文件形式存储持久化消息数据。这里保存了业务的topic消息元数据，路径名是topic名-分区id，比如 channel-topic-0，还有一些__consumer_offsets-xx 的东西，__consumer_offsets 也是一个topic，只不过是 Kafka 内部使用的。

Kafka以Log文件形式存储消息，为防止log过大，通过日志分段(Segment)方式将Log切分成段，比如下面的00000000000000034164 和 00000000000000034191，一个Segment包含一个.log日志文件、.index偏移量索引文件、.timeindex时间戳索引文件，以及其他文件。

每个Segment文件名字由20位数字组成，该数字就是当前文件保存的第一条消息的offset，叫做基准偏移量。

1. .log 是消息的元数据
2. .index 保存消息d的 offset
3. .timeindex 时间戳 offset 索引

执行如下命令可以查看对应文件内容

sh kafka-run-class.sh kafka.tools.DumpLogSegments --files 文件名 --print-data-log

Kafka 存储文件的内容格式是在不断的演进的，所以不同版本可能存储格式不同。本文中文件格式基于Kafka 2.2.0。

4.1 .log

.log 文件保存的是 producer 发送到当前分区的每一条消息内容，也可以使用 grep 来过滤消息。如果消费者发现没有消费到或者重复消费了某条消息，可以使用上面的命令 + grep 关键字 查看消息。

一个Log文件的消息分成很多小段，每段有 baseOffset 和 lastOffset，baseOffset 是本段消息的起始 offset，lastOffset 本段的截止 offset。count 是本段消息数。offset 就是消息的索引。

4.2 .index

.index 文件保存的 log offset 和 Index offset 分别对应 .log 每段消息的起始offset和截止offset 。并不是记录了全部的offset，采用稀疏索引方式存储并检索消息（类似于二分查找）。

系统检索一条消息的过程是：

1. 根据offset定位Segment
2. 根据.index 定位offset所在索引段
3. 去.log查找消息

4.3 .timeindex

带时间戳的索引文件，时间戳跟生产消息时传入的timeStamp有关，具体查找消息过程，还不是很清楚。

公众号：看起来很美（kanqilaihenmei_）