如何在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(五)日志存储
• Linux中Cache内存占用过高清理
• Kafka数据丢失分析

深入理解kafka(五)日志存储

5.1文件目录布局

根目录下有以下5个checkpoint文件: cleaner-offset-checkpoint, log-start-offset-checkpoint, meta.properties, recovery-point-offset-checkpoint, replication-offset-checkpoint

分区目录下有以下目录: 0000xxx.index(偏移量为64位长整形，长度固定为20位), 0000xxx.log, 0000xxx.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的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。

香港服务器选创新互联，2H2G首月10元开通。
创新互联（www.cdcxhl.com）互联网服务提供商,拥有超过10年的服务器租用、服务器托管、云服务器、虚拟主机、网站系统开发经验。专业提供云主机、虚拟主机、域名注册、VPS主机、云服务器、香港云服务器、免备案服务器等。

分享名称：如何在Linux下查看Kafka的相关信息？(linux下查看kafka)
网页链接：http://www.shufengxianlan.com/qtweb/news24/457324.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联