本文主要介绍云音乐对kafka的优化，给生产集群带来了显著的收益，初步统计每年为云音乐节省几百万的成本。本文主要分为两部分，第一部分详细介绍优化kafka性能时遇到的问题及解决过程，第二部分介绍kafka producer的写入原理及batch优化。如果对我们定位问题的过程不太感兴趣的同学，可以直接阅读第二部分。
一、背景
随着云音乐主站流量的增长以及曙光埋点的放量，平台的流量逐渐增长，kafka平台压力越来越大。当前云音乐平台分3个kafka集群：老集群、mirror集群、新集群。老集群和mirror集群历史比较久远，还在使用比较古老的kafka版本。新集群是最近搭建的，使用最新kafka版本，拆分为两个region：default和dawn，其中，default集群用于新流量接入，dawn集群专属用于曙光埋点流量使用。
kafka集群稳定性的两个重要指标为：NetworkProcessorIdle和RequestHandlerIdle，PE同学反馈新kafka集群的这两个指标很低，高峰期在10%以下，集群流量却与老集群相差较大，新kafka集群性能存在严重问题，同时，老集群也不容乐观，idle高峰期15%左右，kafka集群的优化迫在眉睫。
二、问题定位
构建完善监控体系
兵马未动粮草先行，优化要做监控先行，监控指标一来可以指明方向，二来可以验证优化效果，所以，必须要有比较完善的监控才能去做相关优化。由于历史原因，云音乐的kafka集群缺少必要的体系化的监控告警，只有一些比较宏观的监控指标，无法对问题进行多维度深入分析。首先我们需要做的就是构建完善的监控体系。
基于Prometheus构建的监控平台已经是当前业界的标配，使用prometheus有2个最大优点：
1. kafka社区提供一套完善的基于prometheus+grafana的监控系统构建方法，包括指标采集上报和grafana模板，可以很快的构建出kafka的监控平台，详细可参考此文章
2. 利用prometheus提供的PQL语言，可以很灵活的做到多监控维度的下钻和上卷分析，比较容易发现问题
另外，当前网易内部已经有现成的prometheus平台可以直接使用，因此，此方案可谓是不二之选。
基于此，PE同学仅花了一天时间就构建出kafka集群完善的监控系统，在grafana上可以看到集群的各种维度详细指标信息，包括：大盘流量、topic流量、broker流量、副本复制流量、线程idle情况、ISR信息、Consumer指标、Producer指标、Replica指标、Request指标等，这些详细的指标为我们分析问题带来很大的便利。有了监控后，我们开始着手集群的分析和优化工作。
监控指标分析及优化过程
1. 第一次分析及优化：流量均衡
分析
我们对新集群的核心指标进行了分析，并与老集群的指标进行对比，发现新集群存在几个现象：
1. bytesout分布不均，90MB/s ~ 180MB/s之间
2. request handler idle分布不均，20% ~ 80%之间
3. network processor idle都比较高，80%以上
4. msgin分布不均，18K op/s ~
5. Request Queue Size较高，有时会打满到500，Response Queue Size较低
6. 磁盘IOUtil较低
7. 集群单台机器出流量最高在1Gbps左右
基于上述现象我们基本可以得出以下几个结论：
1. 请求处理线程不足
2. 流量分布不均是当前主要矛盾
3. 当前流量并未达到机器硬件的上限
木桶效应很明显，当前集群吞吐量受限于压力比较大的broker，因此，首先我们需要对集群进行流量均衡。
具体指标如下：
优化
PE同学均衡了kafka的bytesOut流量，idle稍微提升，但是效果不太明显。
随后对msgIn流量进行均衡，操作后，idle有了较明显的变化，最低的idle从10%左右涨到了20%左右。
虽然idle上升了不少，但是从绝对值来看，kafka集群还是比较危险，一般idle在30%以下就说明集群已经处于比较高负荷状态，很难应对流量突发情况。所以，优化工作还需继续。
2. 第二次分析及优化：减少请求数
Network Processor Idle较高，Request Handler Idle较低，根据以前经验，当大量meta请求时，会出现此类现象，查看metadata request指标，每秒140左右，量级不大，应该不是这个问题导致。关于这两个idle的相关原理，可以参考kafka网络模型总结
根据第一次优化经验，我们有了一些优化方向，bytesOut影响不大，msgIn影响较大，Request Handler是用来处理所有kafka的请求的（并不只是处理磁盘相关请求），也就是如果能降低request的请求次数，应该就能提升idle值。分析request请求分布情况，发现绝大部分是在produce和fetch请求，因此，接下来的优化方向是如何减轻读写的请求量。
kafka服务端使用zero copy的优化技术，减轻服务端开销，客户端发送的数据块，kafka服务端默认不会做解包（compression.type=producer），也就是通过提升producer端的batch，是可以减轻request请求数的。
分析MsgIn较大的几个topic发现，这些topic的batch size都很低，分布在1-2之间（对应指标messagesinpersec/totalproducerequestspersec），单条记录大小不到1KB，对应写入任务的batch.size使用了默认的16KB，按理来说，batch应该是可以生效的。通过查看相关文章，发现影响producer batch的还有一个参数linger.ms（默认为0），此参数原理可参考文章kafka producer linger.ms和batch.size参数说明。
我们将topic对应的写入任务，配置linger.ms=50，上线后产生了一定的效果，idle从12%左右提升到19%，如下：
查看对应topic的batch size从1提升到4，bytesIn从14MB/s降低到4MB/s，如下：
从上述优化效果来看，基本可以明确优化Producer的batchSize，可以给集群带来很好的效果。随后，我们优化了流量最大的UA日志（占据集群大部分流量）的写入任务配置，集群的整体指标得到了极大的改善，kafka集群整体压力降低60%+，具体指标如下：
总体topic的batchSize从1.67提升到9.71
bytesOut从3.5GB/s降低到1.7GB/s，bytesIn从392MB/s降低到218MB/s
idle从20%增长到90%
3. 优化经验迁移到老集群经历
优化不生效
有了上述优化经验，我们继续将此优化配置应用到老集群。但是，在变更老集群的部分写入任务后，发现调整linger.ms参数后，虽然参数已生效，但是集群指标没有明显变化，对应topic的batchSize指标也没有明显变化。
分析原因并调整生效
搜索了相关资料后，未发现特别有用的相关说明。只能去看kafka producer的源码了，通过分析kafka的源码，发现kafka producer写入record时，如果record中没有指定partitionID，则会随机分配partitionID，放入对应的topic-partition级别的本地缓存batch。同时。另外一个异步Sender线程从缓存batch中拿数据发送到对应broker，此过程会在下面原理章节详细介绍。
弄清楚相关原理后，经过分析发现，我们优化的写入任务并行度120，对应topic有150个分区，写入流量大概40K/s，这样均摊到每个topic-partition的流量大概是：2.2条/s，而我们设置的linger.ms为50，也就是Sender线程在从topic-partition的batch拿数据的时候，只能拿到1条数据，与监控数据也比较吻合。
为了验证分析，我们将任务的并行度从120调整到1，这样每个topic-partition的流量大概是：266条/s。50ms一个batch，大概有13条左右，调整后观察对应topic的batchSize指标，也与我们分析的相符合。
问题原因找到了，接下来需要考虑如何进行优化。只需要将每条记录对应的partition的策略修改即可，起初我们想通过flink的接口FlinkKafkaPartitioner或者kafka自带的接口Partitioner来自定义record对应partition的计算策略（增加计数器，每N条记录变更一次对应的partition或者每N ms时间间隔变更一次）。但是更深入的看了kafka producer源码后，发现2.4版本中，kafka新引入了一种Sticky Partitioner的策略，详情可参考KIP-480，我们将写入任务的kafka client版本升级到2.4使用sticky的partition策略进行测试后发现，batchSize有很大的提升。
将相关原始流量的分发写入任务调优写入batch后，老kafka集群的压力降低了40%+，具体指标如下：
bytesIn从1.04GB/s降低到705MB/s，bytesOut从15.4GB/s降低到9.56GB/s
idle从8%提升到50%+
batchSize从1提升到16
三、原理
通过上述优化过程，我们梳理清楚kafka producer的写入原理，batch的写入不止与linger.ms、batch.size的配置有关，还和partitioner计算record的分区规则有关。
首先，简单介绍下，producer的写入过程，如下图：
kafka producer写入主要分两部分：写入线程和异步sender发送线程。
1. 写入线程
当客户端调用KafkaProducer.send(ProducerRecord)方法，主要有以下四个步骤：
1. 等待topicPartition相关元信息获取（有缓存）
2. 序列化ProducerRecord中key和value（如果有）
3. 如果当前reocord中没有指定partition信息，则调用Partitioner.partition()获取对应的partition
4. 预估当前记录大小并将其追加到相应的topic-partition缓存队列的ProducerBatch中
2. Sender线程
发送线程会一直循环遍历缓存队列，并对数据做相关处理，发送你ProduceRequest请求到对应broker，主要分以下六个步骤：
1. 遍历所有topic-partition队列的队首ProducerBatch
2. 拿到该ProducerBatch做如下判断（图中为了简便并未全部写明）
-
第一个元素的追加时间与当前时间相比，是否超过
linger.ms -
ProducerBatch是否满
-
内存Buffer是否使用完
-
是否close或flush中
3. 如果满足上述的一个条件，则将当前ProducerBatch对应的TopicPartition的leader所在的brokerId放入ready节点队列中
4. 得到可以发送请求的broker的ready节点后，遍历每个brokerId并做如下操作：
-
尽可能多的拿到该brokerId所负责的topicPartition对应的ProducerBatch（当record大小超过
max.request.size时break，为了防止饿死，每次从不同的topicPartition获取） -
将这些ProducerBatch包装成一个ProducerRequest
5. 更新每个batch相关的metric，如topic.{name}.records-per-batch/bytes/compression-rate等
6. 遍历发送每个ProducerRequest到对应的brokerId
3. 服务端
关于服务端的逻辑，此处只做简单介绍，不做深入分析。
服务端收到ProducerRequest后，会将其中的每个ProducerBatch拆出来，使用zero-copy的方式将各topic-partition数据直接append到相应的log-segment文件中，可极大提升服务端的性能。
4. 场景分析
现在我们来详细看下为什么我们将老集群的写入任务linger.ms设置为50时没有效果。
我们写入任务的流量大概是40K/s，写入的topic分区数为150，任务并发为120，如下图所示：
由于老版本（2.4之前）kafkaClient的每个record对应的partition的分配逻辑是随机的，从图中可以很明显看出，落到每个topicParitionQueue写入缓存队列的流量才不过40000/120/150=2.2，而我们设置的linger.ms为40，也就是在40ms内，每个topicPartitionQueue中的数据最多也就一条，发送给每个的batch也就1左右，经过测试也验证了这个结论。
起初我们计划通过flink或者kafka暴露出来的Partitioner接口自定义record对应的partition分配逻辑。但是，还有一个问题比较奇怪，为什么在新集群上却有效果呢？我们深入比较新老集群对应的写入任务并看了kafka相关源码发现，kafka 2.4版本中对record的partition分配策略做了优化。简单来说sticky的分区策略是保证一个topic在一个producerBatch内的所有rocords对应的分区ID保证一样，这样就可以保证一个batch内的数据都放到一个分区里，详情可查看KIP-480。我们将老机器的写入任务使用的kafka client版本也升级到2.4后，发现写入的batch大小从1提升到12左右，也与我们预期相符。
5. 重要监控指标汇总
在此，总结下kafka broker端比较重要的一些监控指标以供后续完善相关告警做参考，具体信息可参考kafka监控指标官方文档。
服务稳定性相关
-
active controller个数：
kafka_controller_kafkacontroller_activecontrollercount，kafka中控节点，正常为1 -
UR分区数：
kafka_server_replicamanager_underreplicatedpartitions，追不上的分区数，正常为0 -
离线分区数：
kafka_controller_kafkacontroller_offlinepartitionscount，未追的分区数（有broker宕机），正常为0 -
网络处理线程idle：
kafka_network_socketserver_networkprocessoravgidlepercent，处理socket请求的线程池空闲值，需保持在至少30%之上 -
请求处理线程idle：
kafka_server_kafkarequesthandlerpool_requesthandleravgidlepercent_total，真正处理request请求的线程池空闲值，需保持在至少30%以上 -
请求的队列大小：
kafka_network_requestchannel_requestqueuesize，默认500，如果经常满说明当前broker的处理慢或压力大 -
ISR shrink频率：
kafka_server_replicamanager_isrshrinkspersec，如果经常出现，则说明经常有副本同步不上leader被移除isr，预示集群存在稳定性问题 -
ISR expand频率：
kafka_server_replicamanager_isrexpandspersec，和shrink相反的指标，表名副本追上leader放进isr的频率
流量相关
-
topic消息client写入条数：
kafka_server_brokertopicmetrics_messagesinpersec -
topic消息client写入大小：
kafka_server_brokertopicmetrics_bytesinpersec -
topic消息client读大小（不包含follower复制）：
kafka_server_brokertopicmetrics_bytesoutpersec -
follower的replication副本复制bytesIn和bytesOut：
ReplicationBytesInPerSec/ReplicationBytesOutPerSec
上述流量指标均可细化到每个topic每个broker级别（无法到partition级别），通过PQL计算几种组合信息，可查看broker/topic/总体级别流量信息。
请求相关
-
producer/client/副本复制请求数：
kafka_network_requestmetrics_requestspersec -
每个topic的producer请求数：
kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec -
每个topic的拉取请求数：
kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec
上述指标，可细化到请求类型（FetchConsumer/FetchFollower/Produce/ListOffset等）、broker和topic级别，也可通过PQL进行各维度组合查看相应指标。
四、结尾
至此，本文完成对kafka的整个优化过程和相关原理描述。后续也可将此优化推广到其他此类场景。优化前需要注意以下四项：
1. 通过监控指标确认写入的topic的batch大小是否为1左右，在kafka-overview中可查看相应指标
2. 确保程序中，写入kafka的records中并未指定key（如果指定key，则默认不会使用sticky分配策略，以确保相同key在一个partition，除非强行指定sticky partitioner）
3. 当linger.ms设置为0时，batch也可能生效，只要数据产生速度大于发送速度即可
4. 当设置linger.ms时，允许数据存在指定的延迟时间
另外，kafka服务还存在一定的优化空间，包括磁盘、网络、内存等场景，这里就做不详细介绍了。
在此，也感谢PE同学：马宏展、童罕，感谢他们提供的帮助，尤其是kafka的监控系统和promethus平台，在优化过程中发挥了重要的作用。
五、参考文章
2. kafka网络模型总结
3. kafka producer linger.ms和batch.size参数说明
4. KIP-480
网友评论