Kafka调优

Posted 2022-05-07 lisin-lee-cooper

topic分区数和消费者数量

topic的分区数大于等于消费者内节点数

存储日志：
每个日志文件的文件名为起始偏移量，因为每个分区的起始偏移量是0，所以，分区的日志文件都以0000000000000000000.log开始；
默认的每个日志文件最大为「log.segment.bytes =102410241024」1G；
为了简化根据offset查找消息，Kafka日志文件名设计为开始的偏移量。

写入日志：
新的消息总是写入到最后的一个日志文件中
该文件如果到达指定的大小（默认为：1GB）时，将滚动到一个新的文件中

读取消息：
根据「offset」首先需要找到存储数据的 segment 段（注意：offset指定分区的全局偏移量）
然后根据这个「全局分区offset」找到相对于文件的「segment段offset」

删除消息：
在Kafka中，消息是会被定期清理的。一次删除一个segment段的日志文件
Kafka的日志管理器，会根据Kafka的配置，来决定哪些文件可以被删除

日志清理：
log.retention.hours
log.retention.minutes
log.retention.ms

其中，优先级为 log.retention.ms > log.retention.minutes > log.retention.hours。
默认情况，在broker中，配置如下：log.retention.hours=168
也就是，默认日志的保留时间为168小时，相当于保留7天。

服务器台数选择
服务器台数= 2 * （生产者峰值生产速率 * 副本 / 100） + 1

磁盘选择
kafka 底层主要是顺序写，固态硬盘和机械硬盘的顺序写速度差不多。
建议选择普通的机械硬盘。

网络选择
网络带宽 = 峰值吞吐量

CPU 选择
num.io.threads = 8 负责写磁盘的线程数，整个参数值要占总核数的 50%。 num.replica.fetchers = 1 副本拉取线程数，这个参数占总核数的 50%的 1/3。
num.network.threads = 3 数据传输线程数，这个参数占总核数的 50%的 2/3。

内存选择
Kafka 内存组成：堆内存 + 页缓存

页缓存：页缓存是 Linux 系统服务器的内存。我们只需要保证 1 个 segment（1g）中25%的数据在内存中就好。
每个节点页缓存大小 =（分区数 * 1g * 25%）/ 节点数。例如 10 个分区，页缓存大小=（10 * 1g * 25%）/ 3 ≈ 1g

如何提升吞吐量

1）生产者提高吞吐量
（1）buffer.memory：发送消息的缓冲区大小，默认值是 32m，可以增加到 64m。 （2）batch.size：默认是 16k。如果 batch 设置太小，会导致频繁网络请求，吞吐量下降；
如果 batch 太大，会导致一条消息需要等待很久才能被发送出去，增加网络延时。
（3）linger.ms，这个值默认是 0，意思就是消息必须立即被发送。一般设置一个 5-100
毫秒。如果 linger.ms 设置的太小，会导致频繁网络请求，吞吐量下降；如果 linger.ms 太长，
会导致一条消息需要等待很久才能被发送出去，增加网络延时。
（4）compression.type：默认是 none，不压缩，但是也可以使用 lz4 压缩，效率还是不
错的，压缩之后可以减小数据量，提升吞吐量，但是会加大 producer 端的 CPU 开销。

2）增加分区

3）消费者提高吞吐量
（1）调整 fetch.max.bytes 大小，默认是 50m。
（2）调整 max.poll.records 大小，默认是 500 条。

4）增加下游消费者处理能力

单条日志大于1M

message.max.bytes 默认 1m，broker 端接收每个批次消息最大值。
max.request.size 默认 1m，生产者发往 broker 每个请求消息最大值。针对 topic
级别设置消息体的大小。
replica.fetch.max.bytes 默认 1m，副本同步数据，每个批次消息最大值。
fetch.max.bytes 默认 Default: 52428800（50 m）。消费者获取服务器端一批
消息最大的字节数。如果服务器端一批次的数据大于该值
（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对
最大值。一批次的大小受 message.max.bytes （broker config）
or max.message.bytes （topic config）影响。