kafka性能参数和压力测试揭秘

转自:http://blog.csdn.net/stark_summer/article/details/50203133

上一篇文章介绍了Kafka在设计上是如何来保证高时效、大吞吐量的，主要的内容集中在底层原理和架构上，属于理论知识范畴。这次我们站在应用和运维的角度，聊一聊集群到位后要怎么才能最好的配置参数和进行测试性能。Kafka的配置详尽且复杂，想要进行全面的性能调优需要掌握大量信息，我也只是通过工作中的一些实战经验来筛选出对集群性能影响最大的几个要点，接下来要阐述的观点也仅限于我所描述的环境下，请大家根据自己的环境适当取舍。

今天的文章分为两大部分，第一部分介绍一下我总结的跟性能有关的一些参数、含义以及调优策略。第二部分会给出一些我自己实践过的测试结果对照组，具体的数值和结果可能因场景、机器、环境而异，但是总体的思路和方法应该是一致的。

在正式进入主题之前，介绍一下本次测试所使用的机器配置：

• 6台物理机，其中三台部署Broker，三台专门用来launch request。

• 每台物理机：24 Processors，189G Memory，2G 单机带宽。

• 执行本次测试时为了能够覆盖到到一些“非常规”的用法，我把Broker的HeapSize设置到了30G。

相关参数介绍

在调试和优化使用Java开发的系统时，第一步肯定绕不开对JVM的调优，Kafka自然也不例外，而JVM调优的重点则是在内存上。

其实Kafka服务本身并不需要很大内存，上篇文章也已经详细介绍过Kafka依赖系统提供的PageCache来满足性能上的要求，利用VisualJVM等工具可以很清晰的分析出Heap Space的占用比例情况。本文中测试时设置30G内存的目的是支持更高的并发，高并发本身就必然会需要更多的内存来支持，同时高并发也意味着SocketBuffer等相关缓存容量会成倍增长。实际使用中，调整内存大小的准则是留给系统尽可能多的空闲内存，Broker本身则是够用就好。

说完了大小设置我们再来聊一下JVM上的垃圾回收器，官方文档里推荐使用最新的G1来代替CMS作为垃圾回收器。不过也明确指出在某些低版本(1.7u21)的JDK上还是会存在一些不稳定的问题。推荐使用的最低版本为JDK 1.7u51。下面是本次试验中Broker的JVM内存配置参数：

-Xms30g -Xmx30g -XX:PermSize=48m -XX:MaxPermSize=48m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35

其实G1早在JDK 1.6u14中就已经作为体验版首次被引入，但是由于最初误宣传需要收费才能使用，和其自身尚不稳定存在Bug等因素，一直等到1.7的后期update版本才逐渐走入我们的视野。

G1相比较于CMS的优势：

• G1是一种适用于服务器端的垃圾回收器，很好的平衡了吞吐量和响应能力。

• 对于内存的划分方法不同，Eden, Survivor, Old区域不再固定，使用内存会更高效。G1通过对内存进行Region的划分，有效避免了内存碎片问题。

• G1可以指定GC时可用于暂停线程的时间（不保证严格遵守）。而CMS并不提供可控选项。

• CMS只有在FullGC之后会重新合并压缩内存，而G1把回收和合并集合在一起。

• CMS只能使用在Old区，在清理Young时一般是配合使用ParNew，而G1可以统一两类分区的回收算法。

G1的适用场景：

• JVM占用内存较大(At least 4G)

• 应用本身频繁申请、释放内存，进而产生大量内存碎片时。

• 对于GC时间较为敏感的应用。

接下来，我们来总结一下Kafka本身可能会对性能产生影响的配置项。

Broker

num.network.threads:3

用于接收并处理网络请求的线程数，默认为3。其内部实现是采用Selector模型。启动一个线程作为Acceptor来负责建立连接，再配合启动num.network.threads个线程来轮流负责从Sockets里读取请求，一般无需改动，除非上下游并发请求量过大。

num.partitions:1

Partition的数量选取也会直接影响到Kafka集群的吞吐性能。例如我写过MapReduce任务从Kafka中读取数据，每个Partition对应一个Mapper去消费数据，如果Partition数量太少，则任务会因为Mapper数不足而非常慢。此外，当Partition数量相对于流入流出的数据量显得较少，或由于业务逻辑和Partition数量没有匹配好造成个别Partition读写数据量大，大量的读写请求集中落在一台或几台机器上时，很容易就会打满NIC的全部流量。不难想象这时不仅这一个Partition的读写会出现性能瓶颈，同Broker上的其他Partition或服务都会陷入一个网络资源匮乏的情况。

queued.max.requests:500

这个参数是指定用于缓存网络请求的队列的最大容量，这个队列达到上限之后将不再接收新请求。一般不会成为瓶颈点，除非I/O性能太差，这时需要配合num.io.threads等配置一同进行调整。

Replica相关配置：

replica.lag.time.max.ms:10000replica.lag.max.messages:4000num.replica.fetchers:1

上篇文章已经简单介绍过上两项配置的含义，这里不再重复，重点说一下第三项配置。对于任意(Broker, Leader)元组，都会有replication.factor-1个Broker作为Replica，在Replica上会启动若干Fetch线程把对应的数据同步到本地，而num.replica.fetchers这个参数是用来控制Fetch线程的数量。

一般来说如果发现Partition的ISR当中只有自己一个Partition，且长时间没有新的Replica增加进来时，就可以考虑适当的增大这个参数加快复制进度。其内部实现上，每个Fetch就对应了一个SimpleConsumer，对于任意一台其他机器上需要Catch-up的Leader，会创建num.replica.fetchers个SimpleConsumer来拉取Log。

当初刚知道这块设计的时候还蛮疑惑的，在Kafka文档开篇的时候就郑重介绍过，同一个ConsumerGroup内的Consumer和Partition在同一时间内必须保证是一对一的消费关系，而这么简单地增加SimpleConsumer就可以提高效率又是什么原因呢？

查看源码，在AbstractFetcherThread.scala里可以看到，Fetch启动的多线程其实就是一个个的SimpleConsumer。

首先，getFetcherId()利用numFetcher来控制FetchId的范围，进而控制Consumer数量。partitionsPerFetcher结构则是一个从Partition到Partition上启动的Fetchers的Mapping。

上面为每个Partition启动的多个Fetcher（也就是SimpleConsumer）之间通过partitionMap: mutable.HashMap[TopicAndPartition, Long]来共享offset，达到并行Fetch数据的目的。因此，通过共享offset既保证了同一时间内Consumer和Partition之间的一对一关系，又允许我们通过增多Fetch线程来提高效率。

default.replication.factor:1

这个参数指新创建一个topic时，默认的Replica数量。当Producer中的 acks!=0 && acks!=1时，Replica的大小可能会导致在Produce数据时的性能表现有很大不同。Replica过少会影响数据的可用性，太多则会白白浪费存储资源，一般建议在2~3为宜。

fetch.purgatory.purge.interval.requests:1000producer.purgatory.purge.interval.requests:1000

首先让我先来介绍一下这个“炼狱”究竟是用来做什么用的。Broker的一项主要工作就是接收并处理网络上发来的Request。这些Request其中有一些是可以立即答复的，那很自然这些Request会被直接回复。另外还有一部分是没办法或者Request自发的要求延时答复（例如发送和接收的Batch）,Broker会把这种Request放入Paurgatory当中，同时每一个加入Purgatory当中的Request还会额外的加入到两个监控对队列：

• WatcherFor队列：用于检查Request是否被满足。

• DelayedQueue队列：用于检测Request是否超时。

Request最终的状态只有一个，就是Complete。请求被满足和超时最终都会被统一的认为是Complete。

目前版本的Purgatory设计上是存在一定缺陷的。Request状态转变为Complete后，并没能立即从Purgatory中移除，而是继续占用资源，因此占用内存累积最终会引发OOM。这种情况一般只会在topic流量较少的情况下触发。更详细的资料可以查阅扩展阅读，在此不做展开。

在实际使用中我也是踩了这个坑过来的，当时的情况是集群新上了一个topic，初期该topic数据很少（Low volume topic），导致那段时间在凌晨3，4点左右会随机有Broker因为OOM挂掉。定位原因后把*.purgatory.purge.interval.requests的配置调整小至100就解决了这个问题。

Kafka的研发团队已经开始着手重新设计Purgatory，力求能够让Request在Complete时立即从Purgatory中移除。

log.flush.interval.ms:Long.MaxValuelog.flush.scheduler.interval.ms:Long.MaxValuelog.flush.interval.messages:Long.MaxValue

Flush相关的配置参数控制着Broker写盘的频率，一般无需改动。如果topic的数据量较小可以考虑减少log.flush.interval.ms和log.flush.interval.messages来强制刷写数据，减少可能由于缓存数据未写盘带来的不一致。

in.insync.replicas:1

这个参数只能在topic层级配置，指定每次Producer写操作至少要保证有多少个在ISR的Replica确认，一般配合request.required.acks使用。要注意，这个参数如果设置的过高可能会大幅降低吞吐量。

compression.codec:none

Message落地时是否采用以及采用何种压缩算法。一般都是把Producer发过来Message直接保存，不再改变压缩方式。

Producer" style="font-weight: 400; font-size: 16px; color: rgb(0, 0, 0); font-family: 'Helvetica Neue', Helvetica, 'Hiragino Sans GB', 'Microsoft YaHei', Arial, sans-serif; line-height: 25.6px;">Producer

buffer.memory:33554432 (32m)

在Producer端用来存放尚未发送出去的Message的缓冲区大小。缓冲区满了之后可以选择阻塞发送或抛出异常，由block.on.buffer.full的配置来决定。

compression.type:none

默认发送不进行压缩，推荐配置一种适合的压缩算法，可以大幅度的减缓网络压力和Broker的存储压力。

linger.ms:0

Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞吐量，而linger.ms则更进一步，这个参数为每次发送增加一些delay，以此来聚合更多的Message。

batch.size:16384

Producer会尝试去把发往同一个Partition的多个Requests进行合并，batch.size指明了一次Batch合并后Requests总大小的上限。如果这个值设置的太小，可能会导致所有的Request都不进行Batch。

acks:1

这个配置可以设定发送消息后是否需要Broker端返回确认。

• 0: 不需要进行确认，速度最快。存在丢失数据的风险。

• 1: 仅需要Leader进行确认，不需要ISR进行确认。是一种效率和安全折中的方式。

• all: 需要ISR中所有的Replica给予接收确认，速度最慢，安全性最高，但是由于ISR可能会缩小到仅包含一个Replica，所以设置参数为all并不能一定避免数据丢失。

注：新老Producer的参数有很大不同，其他配置含义可以对照参考Kafka官方文档。

Consumer

num.consumer.fetchers:1

启动Consumer的个数，适当增加可以提高并发度。

fetch.min.bytes:1

每次Fetch Request至少要拿到多少字节的数据才可以返回。

fetch.wait.max.ms:100

在Fetch Request获取的数据至少达到fetch.min.bytes之前，允许等待的最大时长。对应上面说到的Purgatory中请求的超时时间。

性能测试实战

由于可调整的配置参数较多，为了可以准确的展示不同配置对性能产生的影响，我们每次只调整一个参数，观察对照组结果。测试工具使用Kafka提供的Performance工具ProducerPerformance和ConsumerPerformance。

Producer

Kafka在0.8版本推出了新的Producer Client，较之前版本有极大的性能提升，所以后续的示例无需说明都采用的是新Producer，这里就只给出一组新旧Producer的对照组数据。

其中，Producer的message.size为1024，不压缩，测试时都发送500000条Message。相信大家看过上面结果，就很清楚以后为什么要乖乖地用新设计的Producer来发消息了。

Kafka发布时提供了两个Producer的性能测试工具：

• kafka.tools.ProducerPerformance (Scala)

• org.apache.kafka.clients.tools.ProducerPerformance (Java)

两份工具的大体功能类似。通过Scala版的代码可以很方便的输出CVS文件，通过Patch:1190(https://issues.apache.org/jira/browse/KAFKA-1190)中包含的一个R脚本可以将这个CVS文件结果可视化。

注：如果使用Scala版代码，不建议开启--vary-message-size功能。这个功能使得每次构造消息时都会在内部调用random方法生成随机长度的消息，尤其是在进行压力测试时，构造随机串的消耗累计占比飙高，严重影响发送效率最终致使测试结果失准。

下面，从thread, acks, linger.ms, replica, compression几个主要维度测试了一下Producer的组合性能表现。其中，公共指标如下：

        message.size=1024
        batch.siz=10240
        message.count=50000000

测试结果如下：

注：部分提高了linger.ms的Case效果不明显是由于触发了其他的flush条件。

Consumer

Consumer的测试相对来说就简单很多，毕竟拉取数据时只从Leader读，无论多少Replica都是如此。所以比较关键的参数就聚焦到了fetch.size和thread上。

上述本文给出的参数只是一种参考，适用于我们的集群配置。大家有兴趣可以根据上面提供的方法，在自己的集群上新建独立topic，在实际环境中测试，这样得出的配置才是最适合你的配置。希望大家都能通过上面的方法把自己手头的Kafka调教好，榨干最后一丝性能。

扩展阅读

Request Purgatory潜在引发OOM的问题: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=34839465

Purgatory Redesign: https://cwiki.apache.org/confluence/display/KAFKA/Purgatory+Redesign+Proposal

深入理解G1内存收集器: http://t.cn/RAUulGC

How to choose the number of topics/partitions in a Kafka cluster?: http://www.confluent.io/blog/how-to-choose-the-number-of-topicspartitions-in-a-kafka-cluster

Tips for improving performance of kafka producer: http://ingest.tips/2015/07/19/tips-for-improving-performance-of-kafka-producer/