在Kafka当中数据是以日志的形式存在的
在Kafka当中。数据在磁盘当中的存储:
这些分别对应:
文件名 说明 00000000000000000000.index 索引文件,根据offset查找数据就是通过该索引文件来操作的 00000000000000000000.log 日志数据文件 00000000000000000000.timeindex 时间索引 leader-epoch-checkpoint 持久化每个partition leader对应的 LEO(log end offset、日志文件中下一条待写入消息的offset )为了方便测试观察,新创建一个topic:「test_10m」,该topic每个日志数据文件最大为10M
bin/kafka-topics.sh --create --zookeeper node1.itcast.cn --topic test_10m --replication-factor 2 --partitions 3 --config segment.bytes=10485760
使用之前的生产者程序以往主题中生产数据,可以观察到如下: 每个log文件的大小最大为10M
Kafka 消费者消费数据的速度是非常快的,但如果由于处理 Kafka 消息时,由于有一些外部 IO、或者是产生网络拥堵,就会造成 Kafka 中的数据积压(或称为数据堆积)。如果数据一直积压,会导致数据出来的实时性受到较大影响。
当Kafka出现数据积压问题时,首先要找到数据积压的原因。
以下是在企业中出现数据积压的几个类场景:
数据写入MySQL失败
问题描述: 某日运维人员找到开发人员,说某个 topic 的一个分区发生数据积压,这个topic 非常重要,而且开始有用户投诉。运维非常紧张,赶紧重启了这台机器。重启之后,还是无济于事。
问题分析: 消费这个 topic 的代码比较简单,主要就是消费topic数据,然后进行判断在进行数据库操作。运维通过 kafka-eagle 找到积压的 topic,发现该topic的某个分区积压了几十万条的消息。 最后,通过查看日志发现,由于数据写入到 MySQL 中报错,导致消费分区的 offset 一直没有提交,所以数据积压严重。
因为网络延迟消费失败
问题描述: 基于Kafka开发的系统平稳运行了两个月,突然某天发现某个topic中的消息出现数据积压,大概有几万条消息没有被消费.
问题分析: 通过查看应用程序日志发现,有大量的消费超时失败。后查明原因,因为当天网络抖动,通过查看 Kafka 的消费者超时配置为50ms,随后,将消费的时间修改为 500ms 后问题解决。
Kafka的消息存储在磁盘中,为了控制磁盘占用空间,Kafka需要不断地对过去的一些消息进行清理工作。Kafka 的每个分区都有很多的日志文件,这样也是为了方便进行日志的清理。在Kafka中,提供两种日志清理方式:
在Kafka的broker或topic配置中:
配置项 配置值 说明 log.cleaner.enable true(默认) 开启自动清理日志功能 log.cleanup.policy delete(默认) 删除日志 log.cleanup.policy compaction 压缩日志 log.cleanup.policy delete,compact 同时支持删除、压缩日志删除是以段(segment日志)为单位来进行定期清理的。
定时日志删除任务 Kafka日志管理器中会有一个专门的日志删除任务来定期检测和删除不符合保留条件的日志分段文件,这个周期可以通过broker端参数log.retention.check.interval.ms 来配置,默认值为300,000,即5分钟。当前日志分段的保留策略有3种:
基于时间的保留策略
以下三种配置可以指定如果Kafka中的消息超过指定的阈值,就会将日志进行自动清理
其中,优先级为 log.retention.ms > log.retention.minutes > log.retention.hours。默认情况,在broker中,配置如下: log.retention.hours=168 也就是,默认日志的保留时间为168小时,相当于保留7天。
删除日志分段时:
基于日志大小的保留策略
日志删除任务会检查当前日志的大小是否超过设定的阈值来寻找可删除的日志分段的文件集合。可以通过 broker 端参数 log.retention.bytes 来配置,默认值为 -1,表示无穷大。如果超过该大小,会自动将超出部分删除.
注意:log.retention.bytes 配置的是日志文件的总大小,而不是单个的日志分段的大小,一个日志文件包含多个日志分段
基于日志起始偏移量保留策略
每个segment日志都有它的起始偏移量,如果起始偏移量小于 logStartOffset,那么这些日志文件将会标记为删除.
Log Compaction是默认的日志删除之外的清理过时数据的方式。它会将相同的key对应的数据只保留一个版本。