如何在Linux下查看Kafka的相关信息？ (linux下查看kafka)
Kafka是由Apache软件基金会开发的一款开源的分布式消息系统，它的高吞吐率、高并发性能以及良好的水平扩展性，使得它在数据处理领域中备受青睐。在Linux下使用Kafka，您可能需要查看一些与Kafka相关的信息，包括已安装的Kafka版本、Topic、Partition和Broker等信息。那么，在本文中，我们将介绍如何在Linux下查看Kafka的相关信息。
查看已安装的Kafka版本
在Linux中查看已安装的Kafka版本可以通过以下命令实现：
“`shell
$ kafka-topics.sh –version
“`
该命令将会返回当前Kafka的版本信息，如下：
“`
kafka-topics.sh v2.4.0 (Commit: c57222ae8cd7866d)
“`
查看Topic信息
在Linux中查看Kafka的Topic信息可以通过以下命令实现：
“`shell
$ kafka-topics.sh –describe –topic –zookeeper :
“`
该命令将会返回与指定Topic相关的所有信息，包括Partition数量、Replication Factor、Partition分布情况等，如下：
“`
Topic: PartitionCount:3 ReplicationFactor:2 Configs:
Topic: Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: Partition: 1 Leader: 2 Replicas: 2,3 Isr: 2,3
Topic: Partition: 2 Leader: 3 Replicas: 3,1 Isr: 3,1
“`
其中，“Leader”表示管理该Partition的Broker，而“Replicas”表示Partition的副本集。
查看Broker信息
在Linux中查看Kafka的Broker信息可以通过以下命令实现：
“`shell
$ kafka-broker-api-versions.sh –bootstrap-server : –command-config
“`
该命令将会返回当前Kafka的Broker信息，包括API版本、Broker ID、版本等，如下：
“`
{ version: 4, leader_epoch: -1, attributes: 0 }
… omitted the rest …
( note that the output may include other revisions, you have to look it up).
“`
查看Partition信息
在Linux中查看Kafka的Partition信息可以通过以下命令实现：
“`shell
$ kafka-run-class.sh kafka.tools.GetOffsetShell –broker-list : –topic –time -2
“`
该命令将会返回指定Topic的Partition相关信息，包括Offset信息，如下：
“`
topic_name:0:100
“`
其中，“100”是指该Partition的最新Offset。
查看Consumer Group信息
在Linux中查看Kafka的Consumer Group信息可以通过以下命令实现：
“`shell
$ kafka-consumer-groups.sh –bootstrap-server : –group –describe
“`
该命令将会返回指定Group Name的相关信息，包括当前消费者数量、Offset信息以及Lag信息等，如下：
“`
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test_topic 0 12023 12100 33 YjSIHz94Lua6a-EU0Ti9Xw-1593316979417 /192.168.0.13 YjSIHz94Lua6a-EU0Ti9Xw-1593316979417
“`
其中，“Consumer-ID”是指该消费者当前的ID，而“HOST”是指该消费者所在的IP地址。
相关问题拓展阅读：
深入理解kafka(五)日志存储
5.1文件目录布局
根目录下有以下5个checkpoint文件: cleaner-offset-checkpoint, log-start-offset-checkpoint, meta.properties, recovery-point-offset-checkpoint, replication-offset-checkpoint
分区目录下有以下目录: 0000.index(偏移量为64位长整形，长度固定为20位), 0000.log, 0000.timeindex.
还有可能包含.deleted .cleaned .swap等临时文件, 以及可能的.snapshot .txnindex leader-epoch-checkpoint
5.2日志格式演变
5.2.1 v0版本
kafka0.10.0之前
RECORD_OVERHEAD包括offset(8B)和message size(4B)
RECORD包括:
crc32(4B):crc32校验值
magic(1B):消息版本号0
attributes(1B):消息属性。低3位表示压缩类型：0-NONE 1-GZIP 2-SNAPPY 3-LZ4(0.9.x引入)
key length(4B):表示消息的key的长度。-1代表null
key: 可选
value length(4B):实际消息体的长度。-1代表null
value: 消息体。可以为空，如墓族颤碑消息
5.2.2 v1版本
kafka0.10.0-0.11.0
比v0多了timestamp(8B)字段，表示消息的时间戳
attributes的第4位也被利用起来，0表示timestamp的类型为CreateTime，1表示timestamp的类型为LogAppendTime
timestamp类型由broker端参数log.message.timestamp.type来配置，默认为CreateTime，即采用生产者创建的时间戳
5.2.3 消息压缩
保证端到端的压缩，服务端配置compression.type，默认为”producer”,表示保留生产者使用的压缩方式，还可以配置为”gzip”,”snappy”,”lz4″
多条消息压缩至value字段，以提高压缩率
5.2.4 变长字段
变长整缺隐形(Varints):每一个字节都有一个位于更高位的m位(most significant bit),除了最后一个字节为1，其余都为0，字节倒序排列
为了使编码更加高效，Varints使用ZigZag编码:sint32对应 (n>31) sint64对应 (n>63)
5.2.5 v2版本
Record Batch
first offset:
length:
partition leader epoch:
magic:固定为2
attributes:两个字节。低3位表示压缩格式，第4位表示时间戳类型，第5位表示事务(0-非事务1-事务)，第6位控制消息(0-非控制1控制)
first timestamp:
max timestamp:
producer id:
producer epoch:
first sequence:
records count:
v2版本的消息去掉了crc字段，另外增加了length(消息总长度)、timestamp delta(时间戳增量)、offset delta(位移增伏穗厅量)和headers信息，并且弃用了attributes
Record
length:
attributes:弃用，但仍占据1B
timestamp delta:
offset delta:
headers:
5.3日志索引
稀疏索引(sparse index):每当写入一定量(broker端参数log.index.interval.bytes指定，默认为4096B),偏移量索引文件和时间索引文件分别对应一个索引项
日志段切分策略：
1.大小超过broker端参数log.segment.bytes配置的值，默认为(1GB)
2.当前日志段消息的更大时间戳与当前系统的时间戳差值大于log.roll.ms或者log.roll.hours，ms优先级高，默认log.roll.hours=168(7天)
3.索引文件或者时间戳索引文件的大小大于log.index.size.max.bytes配置的值，默认为(10MB)
4.偏移量差值(offset-baseOffset)>Integer.MAX_VALUE
5.3.1 偏移量索引
每个索引项占用8个字节，分为两个部分：1.relativeOffset相对偏移量(4B) 2.position物理地址(4B)
使用kafka-dump-log.sh脚本来解析.index文件(包括.timeindex、.snapshot、.txnindex等文件)，如下:
bin/kafka-dump-log.sh –files /tmp/kafka-logs/topicId-0/00……00.index
如果broker端参数log.index.size.max.bytes不是8的倍数，内部会自动转换为8的倍数
5.3.2 时间戳索引
每个索引项占用12个字节，分为两个部分：1.timestamp当前日志分段的更大时间戳(12B) 2.relativeOffset时间戳对应的相对偏移量(4B)
如果broker端参数log.index.size.max.bytes不是12的倍数，内部会自动转换为12的倍数
5.4日志清理
日志清理策略可以控制到主题级别
5.4.1 日志删除
broker端参数log.cleanup.policy设置为delete(默认为delete)
检测周期broker端参数log.retention.check.interval.ms=300000(默认5分钟)
1.基于时间
broker端参数log.retention.hours,log.retention.minutes,log.retention.ms,优先级ms>minutes>hours
删除时先增加.delete后缀，延迟删除根据file.delete.delay.ms(默认60000)配置
2.基于日志大小
日志总大小为broker端参数log.retention.bytes(默认为-1，表示无穷大)
日志段大小为broker端参数log.segment.bytes(默认为，1GB)
3.基于日志起始偏移量
DeleteRecordRequest请求
1.KafkaAdminClient的deleteRecord()
2.kafka-delete-record.sh脚本
5.4.2 日志压缩
broker端参数log.cleanup.policy设置为compact，且log.cleaner.enable设置为true(默认为true)
5.5磁盘存储
相关测试：一个由6块7200r/min的RAID-5阵列组成的磁盘簇的线性写入600MB/s，随机写入100KB/s，随机内存写入400MB/s，线性内存3.6GB/s
5.5.1 页缓存
Linux操作系统的vm.dirty_background_ratio参数用来指定脏页数量达到系统的百分比之后就触发pdflush/flush/kdmflush，一般小于10，不建议为0
vm.dirty_ratio表示脏页百分比之后刷盘，但是阻塞新IO请求
kafka同样提供同步刷盘及间断性强制刷盘(fsync)功能，可以通过log.flush.interval.messages、log.flush.interval.ms等参数来控制
kafka不建议使用swap分区，vm.swappiness参数上限为100，下限为0，建议设置为1
5.5.2 磁盘I/O流程
一般磁盘IO的场景有以下4种：
1.用户调用标准C库进行IO操作，数据流为：应用程序Buffer->C库标准IOBuffer->文件系统也缓存->通过具体文件系统到磁盘
2.用户调用文件IO，数据流为：应用程序Buffer->文件系统也缓存->通过具体文件系统到磁盘
3.用户打开文件时使用O_DIRECT，绕过页缓存直接读写磁盘
4.用户使用类似dd工具，并使用direct参数，绕过系统cache与文件系统直接读写磁盘
Linux系统中IO调度策略有4种：
1.NOOP：no operation
2.CFQ
3.DEADLINE
4.ANTICIPATORY
5.5.3 零拷贝
指数据直接从磁盘文件复制到网卡设备中，不需要经应用程序
对linux而言依赖于底层的sendfile()
对java而言，FileChannal.transferTo()的底层实现就是sendfile()
Linux中Cache内存占用过高清理
在Linux中每次用free查看的时候，发现free的空间都只有500M左右。同样的环境32G只剩下这点，64G的也只剩下这么一点。后来发现都被Cache占用了，因为服务器上运行了Kafka环境，每周的日志文件都有一二百G的，估计就是他占用了page cache吧。
Free中的buffer和cache：（它们都是占用内存）：
buffer : 作为buffer cache的内存，是块设备的读写缓冲区
cache: 作为page cache的内存, 文件系统的cache
如果乱链握 cache 的值很大，说明cache住的文件数很多。如果频繁访问到的文件都能被cache住，那么磁盘的读IO bi会非常小。
Linux内核会在内存将要唤大耗尽的时候，触发内存回收的工作，以便释放出内存给急需内存的进程使用。也可以用动释放，释放的时候需要对cache中的数据跟对应文件中的数据一致。
释放的方式有下以几种
一般情况下释放pagecache就可以了。这样可以写一个sh脚本来在服务器空闲的时候定时执行
使用哗庆crontab来设置定时任务，如每天4点开始清理
本文参考：
Kafka数据丢失分析
Kafka存在丢消息的问题，消息丢失会发生在Broker，Producer和Consumer三种。
Broker丢失消息是由于Kafka本身的原因造成的，kafka为了得到更高的性能和吞吐量，将数据异步批量的存储在磁盘中。消息的刷盘过程，为了提高性能，减少刷盘次数，kafka采用了批量刷盘的做法。即，按照一定的消息量，和时间间隔进行刷盘。这种机制也是由于linux操作系统决定的。将数据存储到linux操作系统种，会先存储到页缓存（Page cache）中，按照时间或者其他条件进行刷盘（从page cache到file），或者通过fsync命令强制刷盘。数据在page cache中时，如果系统挂掉，数据会丢失。
Broker在linux服务器上高速读写以及同步到Replica
上图简述了broker写数据以及同步的一个过程。broker写数据只写到PageCache中，而pageCache位于内存。这部分数据在断电后是会丢失的。pageCache的数据通过linux的flusher程序进行刷盘。刷盘触发条件有三：
Broker配置刷盘机制，是通过调用fsync函数接管了刷盘动作。从单个Broker来看，pageCache的数据会丢失。
Kafka没有提供同步刷盘的方式。同步刷盘在RocketMQ中有实现，实现原理是将异步刷盘的流程进行阻塞，等待响应，类似ajax的callback或者是java的future。下面是一段rocketmq的源码。
也就是说，理论上，要完全让kafka保证单个broker不丢失消息是做不到的，只能通过调整刷盘机制的参数缓解该情况。比如，减少刷盘间隔，减少刷盘数据量大小。时间越短，性能越差，可靠性越好（尽可能可靠）。这是一个选择题。
为了解决该问题，kafka通过producer和broker协同处理单个broker丢失参数的情况。一旦producer发现broker消息丢失，即可自动进行retry。除非retry次数超过阀值乎橘（可配置），消息才会丢失。此时需要生产者客户端手动处理该情况。那么producer是如何检测到数据丢失的呢？是通过ack机制，类似于http的三次握手的方式。
以上的引用是kafka官方对于参数 acks 的解释（在老版本中，该参数是 request.required.acks ）。
上面第三点提到了ISR的列表的follower，需要配合另一个参数才能更好的保证ack的有效性。ISR是Broker维护的一个“可靠的follower列表”，in-sync Replica列表，broker的配置包含一个参数： min.insync.replicas 。该参数表示ISR中最少的副本数。如果不设置该值，ISR中的follower列表可能为空。此时相当于acks=1。
如上图中：
性能依次递减，可靠性依次升高。
Producer丢失消息，发生在生产者客户端。
为了提升效率，减少IO，producer在发送数据时可以将多个请求进行合并后发送。被合并的请求咋发送一线缓存在本地buffer中。缓存的方式和前文提到的刷盘类似，producer可以将请求打包成“块”或者按照时间间隔，将buffer中的数据发出。通过buffer我们可以将生产者改造为异步的方式，而这可以提升我们的发送效率。
但是，buffer中的数据就是危险的。在正常情况下，客户端的异步调用可以通过callback来处理消息发送失败或者超时的情况，但是，一旦producer被非法的停止了岁谈团，那么buffer中的数据将丢失，broker将无法收到该部分数据。又或者，当Producer客户端内存不够时，如果采取的策略是丢弃消息（另一种策略是block阻塞），消息也会被丢失。抑或，消息产生（异步产生）过快，导致挂起线程过多，内存不足，侍戚导致程序崩溃，消息丢失。
producer
根据上图，可以想到几个解决的思路：
Consumer消费消息有下面几个步骤：
Consumer的消费方式主要分为两种：
上面的示例是自动提交的例子。如果此时，insertIntoDB(record)发生异常，消息将会出现丢失。接下来是手动提交的例子：
将提交类型改为手动以后，可以保证消息“至少被消费一次”(at least once)。但此时可能出现重复消费的情况，重复消费不属于本篇讨论范围。
上面两个例子，是直接使用Consumer的High level API，客户端对于offset等控制是透明的。也可以采用Low level API的方式，手动控制offset，也可以保证消息不丢，不过会更加复杂。
关于linux下查看kafka的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。
