作为一个 Kafka 初学者,需要快速成长,承担维护公司 Kafka 的重任,对 Kafka 的学习,我按照三步走策略:
- 阅读 Kafka 相关书籍
- 从运维实战的角度学习 Kafka
- 阅读源码,体系化,精细化掌握其实现原理
本文属于学习的第二阶段:[从运维实战的角度学习 Kafka],重点学习 Kafka 的主题,通过运维命令创建、更新主题,从 Topic 的可运维属性,了解 Topic 在 Kafka 内部的运作机制。
Kafka topic 运维命令的基本使用
Kafka 提供了 kafka-topics 脚步用来创建、修改、删除、查询 topic,位于${kafka_home}/bin/kafka-topics.sh,其中 kafka_home 表示 Kafka 的安装目录。
一些不那么直观的选项进行单独介绍。
--replica-assignment
收到指定副本数量和分区信息,该参数不能和--partitions、--replication-factor 同时使用。
![](https://img.haomeiwen.com/i11870810/4b324e41ad89c06a.png)
其格式为:每一个逗号表示一个分区的配置,每一个分区分布的 broker 用冒号隔开。
--replication-factor 0:1,1:2,0:2 表示的含义是什么呢?
分区数量为 3 个,其中分区 0(p0)分布在 broker 0 和 1 上,分区 1(p1)分布在 broker 1,2 上,分区 2(p2)分布在 broker 0 与 2 上。从而推出分区数量为 3,副本因子为 2,每一个分区的第一个 broker 为 Leader,其演示效果如下:
![](https://img.haomeiwen.com/i11870810/6b9df80cc925a6f0.png)
Kafka Topic 配置项详解
通过 kafka-topics 脚本在创建 topic 时可通过--config 选项来定制化 topic 的属性,接下来试图从这些属性来探究 Kafka 背后的运作机制。
- cleanup.policy
数据文件清除机制,支持 Broker 全局配置,Topic 定制化制定,可选策略:delete、compact,默认值为 delete。Kafka 提供了数据段压缩的功能,按照相同 Key 只保留最新 Key 的策略,减少数据段大小,系统主题__consumer_offsets(用于存储消息进度的主题)其清除策略就是 compact。
- compression.type
压缩类型,Kafka 目前支持的压缩算法:gzip,snappy,lz4,zstd,还支持如下两个配置:
- uncompressed
不开启压缩
- producer
由发送方指定压缩算法,客户端的可选值为 gzip,snappy,lz4,zstd。
数据进行压缩,能节省网络带宽与存储空间,但会增加 CPU 的性能,故最佳实践:Broker 服务端不配置压缩算法,由发送方指定,在发送方进行压缩,服务端原封不动进行存储,并且在消费端解压缩。
- delete.retention.ms
如果 cleanup.policy 策略为 compact 时,针对消息体为 null 的消息,Kafka 会认为对其进行压缩没有意义,立马删除也太草率,故 Kafka 引入了该参数,用来设置这些 body 为 null 的消息,在一次压缩执行后,多久后可被删除,默认值为 24h。
- file.delete.delay.ms
文件在删除时延迟时间,默认为 60s,Kafka 中可以支持按 topic 删除日志文件(数据文件),执行删除之前,首先会将该 topic 下的分区文件重名为*.deleted,等待 file.delete.delay.ms 才从文件系统中删除。
- flush.messages
按消息条数设置刷盘频率,如果设置为 1 表示每写一条消息就触发一次刷盘,默认值为 Long.MaxValue,在大部分场景官方不建议设置该值,直接利用操作系统的刷盘机制即可,Kafka 希望通过副本机制能保证数据的持久可靠存储。
- flush.ms
按时间间隔设置刷盘频率,默认为 Long.MaxValue,Kafka 希望借助操作系统的刷盘机制,数据可靠性通过副本机制来保证。(副本机制其实无法保证同机房断电带来的数据丢失)
- index.interval.bytes
索引文件的密度,Kafka 并不会为每一条消息(消息偏移量)建立索引,而是每隔一定间隔,建立一条索引。该参数就是设置其间隔,默认为 4096 个字节。
- max.message.bytes
一次消息发送(Batch)允许的最大字节数量,默认为 1000000,约等于 1M。
- message.downconversion.enable
是否开启消息格式的自动转化,如果设置为 false,Broker 不会执行消息格式转化,将不兼容老的客户端消费消息。
- message.format.version
可以指定该主题按特定版本的 API 版本所对应的存储格式进行存储。
- message.timestamp.type
设置消息中存储的时间戳的获取方式,可选值:
- CreateTime
消息在客户端的创建时间
- LogAppendTime
Broker 服务端接收到的时间,默认为 CreateTime。
- message.timestamp.difference.max.ms
当 message.timestamp.type 设置为 CreateTime 时,允许 Broker 端时间与消息创建时间戳最大的差值,如果超过该参数设置的阔值,Broker 会拒绝存储该消息,默认为:Long.MaxValue,表示不开启开机制。
- min.cleanable.dirty.ratio
控制可压缩的脏数据比例,默认为 0.5d,如果一个文件中"脏数据"(未被压缩的数据)低于该阔值,将不继续对该文件进行压缩,该方法生效的条件为 cleanup.policy 设置为 compact。
- min.compaction.lag.ms
设置一条消息进入到 Broker 后多久之内不能被 compact,默认为 0,表示不启用该特性,该方法生效的条件为 cleanup.policy 设置为 compact。
- min.insync.replicas
如果客户端在消息发送时将 ack 设置为 all,该参数指定必须至少多少个副本写入成功,才能向客户端返回成功,默认为 1,这个是一个兜底配置,all 的含义表示在 ISR 中的副本必须全部写入成功。
- preallocate
是否开启预热文件(提前创建文件),默认为 false。
- retention.bytes
一个日志分区保留的最大字节数,默认为-1,表示不限制。
- retention.ms
一个日志分区允许保留的最大时长,默认保留 7d。
- segment.bytes
一个日志段的大小,默认为 1G。
- segment.index.bytes
一个日志段索引文件的大小,默认为 10M。
- segment.jitter.ms
段滚动的最大随机差。
- segment.ms
Kafka 强制滚动一个段的间隔时间,及时该段并未全部填满消息,默认值为 7d
- unclean.leader.election.enable
是否允许不在 ISR 中副本在没有 ISR 副本选择之后竞争成为 Leader,这样做有可能丢数据,默认为 false。
总结
本文从运维命令开始学习,从使用运维层面全面了解 Topic,从而窥探其 Kafka 内部一些重要特性,为后续从源码角度研究其实现打下坚实基础。
本文的最后给出一个分区数量为 3,副本因子为 3 的 topic 分区图来结束本文的讲解。
![](https://img.haomeiwen.com/i11870810/04101c9a46cb4ca7.png)
网友评论