Flink-Kafka性能压测全记录

1.压测方案

1.1 压测目的

本次性能测试在正式环境下单台服务器上Kafka处理MQ消息能力进行压力测试。测试包括对Kafka写入MQ消息和消费MQ消息进行压力测试,根据不同量级的消息处理结果,评估Kafka的处理性能是否满足项目需求(该项目期望Kafka能够处理上亿级别的MQ消息)。

1.2 测试范围及方法

1.2.1 测试范围概述

测试使用Kafka自带的测试脚本,通过命令对Kafka发起写入MQ消息和Kafka消费MQ消息的请求。模拟不同数量级的消息队列的消息写入和消费场景,根据Kafka的处理结果,评估Kafka是否满足处理亿级以上的消息的能力。

1.2.2测试方法

  • 测试目的:

验证单台服务器上Kafka写入消息和消费消息的能力,根据测试结果评估当前Kafka集群模式是否满足上亿级别的消息处理能力。

  • 测试方法

在服务器上使用Kafka自带的测试脚本,模拟1y级别的消息写入以及读取请求,查看Kafka处理不同数量级的消息数时的处理能力,包括每秒生成消息数、吞吐量、消息延迟时间。

Kafka消息写入创建的topic命名为test_kafka_throughout,Kafka消费读取的topic也是该topic;使用命令发起消费该topic的请求,针对不同的测试指标,本次我们采用固定其他值,动态变化测量值的方式来进行,具体使用脚本为kafka自带的测试脚本,分别为kafka bin目录下的kafka-producer-perf-test.sh和kafka-consumer-perf-test.sh;通过测试来查看Kafka消费不同数量级别的消息时的处理能力。

  • 准备工作

测试之前,我们需要先用linux命令去测试磁盘的读写速度,具体命令如下:

1.测试IO读
    hdparm -t --direct /dev/sda3
    IO读用上面的命令测试即可,不过 hdparm 这个工具需要自己安装,而且需要root用户去执行;
2.测试IO写
    sync;/usr/bin/time -p bash -c "(dd if=/dev/zero of=test.dd  bs=1M count=20000)"
测试结论:
    1.dd测试出的读速度和hdparm 是存在区别的;
    2.通过 bs 选项 设置不通的读写块大小测试(默认512字节,测试使用1M);
    3.可以看出 dd 的测出的速度与读写块的大小有关系,还可能受到系统中有IO读写的进程的影响;
    4.hdparm的测试原理可能是和dd的测试方法存在差别;
    
整体上看,IO的实际测试速度是受到很多因素影响的,包括读写的方式(随机还是顺序,hdparm和dd测试是都是采用顺序读写)、缓存机制、测试的取样等等。
所以不太可能得到一个确定的值(相同的命令行多次测试也不一样,不过差别要小些),以上的方法中读测试还是推荐使用hdparm。
以上的数据虽然存在偏差,但还是能大体分析出机器的IO性能。只是需要明确,这些测试值是在什么样的环境下获得的。
3.测试结果
    1.磁盘cache读7471m/s;
    2.disk读163m/s;
    3.IO写125m/s;
    4.IO读206m/s;
经过测试,我们拿到的磁盘读应该在163m/s-206m/s之间,而写速度是163m/s。后续评测我们以该磁盘测试为基准来核定。

1.3 测试环境

2.kafka参数

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

其实Kafka服务本身并不需要很大内存,其依赖的是系统提供的PageCache来满足性能上的要求,本次测试时设置30G内存的目的是支持更高的并发,高并发本身就必然会需要更多的内存来支持,同时高并发也意味着SocketBuffer等相关缓存容量会成倍增长。实际使用中,调整内存大小的准则是留给系统尽可能多的空闲内存,Broker本身则是够用就好。

JVM上的垃圾回收器,官方文档里推荐使用最新的G1来代替CMS作为垃圾回收器。为了稳定性问题,本次我们使用jdk8以上的版本,我们本次使用G1回收器的原因如下:

  • G1是一种适用于服务器端的垃圾回收器,很好的平衡了吞吐量和响应能力;
  • 对于内存的划分方法不同,Eden, Survivor, Old区域不再固定,使用内存会更高效。G1通过对内存进行Region的划分,有效避免了内存碎片问题;
  • G1可以指定GC时可用于暂停线程的时间(不保证严格遵守)。而CMS并不提供可控选项;
  • CMS只有在FullGC之后会重新合并压缩内存,而G1把回收和合并集合在一起;
  • CMS只能使用在Old区,在清理Young时一般是配合使用ParNew,而G1可以统一两类分区的回收算法。

其使用场景如下:

  • JVM占用内存较大(At least 4G);
  • 应用本身频繁申请、释放内存,进而产生大量内存碎片时;
  • 对于GC时间较为敏感的应用。    首先,我们设置JVM配置为:
-Xmx6G -Xms6G -XX:MMetaspaceSize=96m -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16m -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80
后续在broker的相关参数测试完成后,保持其在最优的参数下,再来针对我们的服务器和kafka的页缓存及jvm堆内存以及策略等进行测试。

2.1 Producer相关参数

   我们在producer涉及到性能的关键因素可能会存在如下几个:

  • thread:我们测试时的单机线程数;
  • bath-size:我们所处理的数据批次大小;
  • ack:主从同步策略我们在生产消息时特别需要注意,是follower收到后返回还是只是leader收到后返回,这对于我们的吞吐量影响颇大;
  • message-size:单条消息的大小,要在producer和broker中设置一个阈值,且它的大小范围对吞吐量也有影响;
  • compression-codec:压缩方式,目前我们有不压缩,gzip,snappy,lz4四种方式;
  • partition:分区数,主要是和线程复合来测试;
  • replication:副本数;
  • througout:我们所需要的吞吐量,单位时间内处理消息的数量,可能对我们处理消息的延迟有影响;
  • linger.ms:两次发送时间间隔,满足后刷一次数据。

2.2 Consumer相关参数

  • thread:我们测试时的单机线程数;
  • fetch-size:抓取数据量;
  • partition:分区数,主要是和线程复合来测试;
  • replication:副本数;
  • througout:我们所需要的吞吐量,单位时间内处理消息的数量,可能对我们处理消息的延迟有影响;

2.3 Broker相关参数

  • num.replica.fetchers:副本抓取的相应参数,如果发生ISR频繁进出的情况或follower无法追上leader的情况则适当增加该值,==但通常不要超过CPU核数+1;==
  • num.io.threads:broker处理磁盘IO的线程数,主要进行磁盘io操作,高峰期可能有些io等待,因此配置需要大些。==建议配置线程数量为cpu核数2倍,最大不超过3倍;==
  • num.network.threads:broker处理消息的最大线程数,和我们生产消费的thread很类似主要处理网络io,读写缓冲区数据,基本没有io等待,==建议配置线程数量为cpu核数加1;==
  • log.flush.interval.messages:每当producer写入多少条消息时,刷数据到磁盘;
  • log.flush.interval.ms:每隔多长时间,刷数据到磁盘;

4.测试过程

4.1 producer测试

4.1.1 bath-size

  • 测试脚本
./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=10000   --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000   --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=40000   --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=60000   --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=80000   --throughput 30000
  • 测试结果
  • 测试结论

测试中通过我们增加batch-size的大小,我们可以发现在消息未压缩的前提下,20000条一批次之后吞吐稳定在30000条/s,而数据量在19.65M/s。

4.1.2 ack

  • 测试脚本
./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 4096  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=0   --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 4096  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=1  --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 4096  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=-1  --throughput 30000
  • 测试结果
  • 测试结论

测试中通过我们使用不同的ack策略,我们可以发现在消息未压缩的前提下,不响应速度最快,其次是leader响应,而follower响应吞吐只有其25%左右,在主从同步策略上要根据数据量还有我们的数据稳定性结合来考量。

4.1.3 message-size

  • 测试脚本
./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=-1  --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 454  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=-1  --throughput 30000
  • 测试结果
  • 测试结论

测试中通过我们使用两种不同的消息大小,发现在消息未压缩的前提下且其他参数一致的情况下,687字节的吞吐量是要优于454字节的,目前我们的两种消息为此大小,测试中发现当消息大小为4k时效果最优,这点可以在后续实践中再去证实

4.1.4 compression-codec

  • 测试脚本
./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=1 compression.type=none  --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=1 compression.type=gzip --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=1  compression.type=snappy --throughput 30000./kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=20000 acks=1  compression.type=lz4 --throughput 30000
  • 测试结果1

在batch-size为2w且并发量在3w时,可以看出来不压缩的吞吐量最好,其他的基本相差不大。

  • 测试结果2

我们在后续测试中发现,在batch-size为100w且并发量在10w时,可以看出来snappy和lz4的吞吐量上升幅度明显,而gzip由于压缩的费时其吞吐最差,不压缩的在本测试中的吞吐次之。

  • 测试结果3

我们在后续测试中发现,在batch-size为100w且并发量在20w时,lz4的吞吐量优势明显达到19w/s,snappy次之为12.8w/s,而gzip由于压缩的费时其吞吐最差基本在5.8w/s,不压缩的在本测试中的吞吐也能达到11w/s。

  • 测试结果4

在batch-size为100w且并发量在50w时,lz4的吞吐量优势明显达到31.3w/s,snappy次之为16.1w/s,而gzip由于压缩的费时其吞吐最差基本在5.3w/s,不压缩的在本测试中的吞吐也能达到9.3w/s。

  • 测试结果5

在batch-size为100w且并发量在60w时,lz4的吞吐达到37.5w/s,snappy此时下降到10.8w/s,而gzip由于压缩的费时其吞吐最差基本在5.4w/s,不压缩的在本测试中的吞吐为9.4w/s。

  • 测试结果6

在batch-size为100w且并发量在70w时,lz4的吞吐量下降到达到27.2w/s,snappy次之为13.9w/s,而gzip则继续保持在5.8w/s,不压缩则下降到7.1w/s。

  • 测试结果7

测试单副本单分区下的各压缩的吞吐量:

我们这次使用1个分区1个副本的主题,测试中通过我们使用不同的压缩格式,在其他参数一致的情况下,在并发和batch-size增大到60w和100w的情况下,lz4达到最好的吞吐21.2w/s,而普通不压缩的方式则维持在6.7w/s。

  • 测试结论

本次测试对数据的存储块大小未测,但在之前的测试中发现压缩以及解压的情况也是lz4算法最优,==lz4压缩最大时可以达到30w+/s的吞吐,而不压缩为12w/s,snappy最大为16w/s,gzip最大为5.8w/s==;故后续生产消息时建议采用lz4压缩,不仅可以节省磁盘,也可以大幅度增加我们的吞吐。

4.1.5 partition

  • 测试脚本
1、创建topic
/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf11  --partitions 1 --replication-factor 1/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf8 --partitions 2 --replication-factor 1/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf16 --partitions 3 --replication-factor 1/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka1 --topic test_kafka_perf24 --partitions 4 --replication-factor 1/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf32 --partitions 5 --replication-factor 12、生产数据
/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 10240  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=80000 acks=1  compression.type=lz4 --throughput 290003、初步结论
分区数越多,单线程消费者吞吐率越小。随着更多的broker线程和磁盘开始扫描不同的分区,吞吐量开始显著增加。但是,一旦使用了所有broker线程和磁盘,添加额外的分区没有任何效果。
  • 测试结果
  • 测试结论

在我们的broker线程小于partiton数时,随着线程增多,吞吐上升,而在两者对等时,达到最优,后续基本稳定,但是由于网络和磁盘的问题可能会有一些起伏。

4.1.6 replication

  • 测试脚本
1、创建topic

/bin/kafka-topics.sh--create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep2 --partitions 1 --replication-factor 1/bin/kafka-topics.sh--create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep3 --partitions 1 --replication-factor 2/bin/kafka-topics.sh--create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep24 --partitions 1 --replication-factor 32、生成数据
/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 5000003、初步结论
备份数越多,吞吐率越低。
  • 测试结果
  • 测试结论

Replication是我们对不同partition所做的副本,它的大小会在ISR中显示,为了保证数据的安全性,ISR中掉出的版本应该保持在1,所以此处我们从replica为2开始测试。在ack不同时,其数量的多少会对性能造成线性的影响,数量过少会影响数据的可用性,太多则会白白浪费存储资源,一般建议在2~4为宜,我们设置为3个,既能保障数据的高可用,又避免了浪费过多的存储资源。

4.1.7 throughout/IO

  • 测试脚本:
/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 10240  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 10000/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 10240  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 30000/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 10240  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 50000/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 10240  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 70000/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 100000000 --record-size 10240  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 100000
  • 测试结果
  • 测试结论

在主题是一个分区和一个副本时,我们看到在并发50w以下时,随着并发数增大,吞吐上升,但是在50w以后时,可以看出并发增大反而吞吐降低了,这是因为IO的限制,在高并发的情况下,产生了阻塞而导致。

4.2 consumer测试

4.2.1 thread

  • 测试脚本
./kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf1 --fetch-size 1048576  --messages 100000  --threads 1 --hide-header --num-fetch-threads 1./kafka-consumer-perf-test.sh --zookeeper  110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf1 --fetch-size 1048576  --messages 100000  --threads 4 --hide-header  --num-fetch-threads 1./kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf1 --fetch-size 1048576  --messages 100000  --threads 7 --hide-header  --num-fetch-threads 1./kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_perf1 --fetch-size 1048576  --messages 100000  --threads 10 --hide-header  --num-fetch-threads 1
  • 测试结果
  • 测试结论

在threads为4时,消费速度最好达到24.1w/s,而后续慢慢平稳。

4.2.2 fetch-size

  • 测试脚
./bin/kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic s1 --messages 1000000 --fetch-size  1000  --threads 3 --hide-header./bin/kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka--topic s1 --messages 1000000 --fetch-size  2000  --threads 3 --hide-header ./bin/kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic s1 --messages 1000000 --fetch-size  5000  --threads 3  --hide-header./bin/kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic s1 --messages 1000000 --fetch-size  10000  --threads 3 --hide-header./bin/kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic s1 --messages 1000000 --fetch-size  15000  --threads 3 --hide-header./bin/kafka-consumer-perf-test.sh --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic s1 --messages 1000000 --fetch-size  20000  --threads 3 --hide-header
  • ‍测试结果
  • 测试结论

4.2.3 partition

  • 测试脚本
1、创建topic/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka  --topic test_kafka_part3  --partitions 3 --replication-factor 1 /bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka  --topic test_kafka_part5  --partitions 5 --replication-factor 1/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_part7  --partitions 7 --replication-factor 1/bin/kafka-topics.sh --create --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_part7  --partitions 9 --replication-factor 12、生成数据/bin/kafka-producer-perf-test.sh  --topic test_kafka_perf1 --num-records 10000000 --record-size 687  --producer-props   bootstrap.servers=10.240.1.134:9092,10.240.1.143:9092,10.240.1.146:9092  batch.size=1000000 acks=1  compression.type=lz4 --throughput 5000003、消费数据./kafka-consumer-perf-test.sh --broker-list localhost:9092 --zookeeper 110.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka  --topic test_kafka_part --messages 10000000 --fetch-size  10240  --threads 5 --hide-header   --num-fetch-threads 14、初步结论分区数越多,单线程消费者吞吐率越小。随着更多的broker线程和磁盘开始扫描不同的分区,吞吐量开始显著增加。但是,一旦使用了所有broker线程和磁盘,添加额外的分区没有任何效果。
  • 测试结果
  • 测试结论

分区数在kafka中和处理的线程数有一定的关系,当thread小于partition数时,那么可能存在一个thread消费两个partition,而==两者一样或者说thread大于partition时,实际是一一对应关系==。

4.2.4 replication

  • 测试脚本
1、创建topic
/bin/kafka-topics.sh--create --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep2 --partitions 5 --replication-factor 2/bin/kafka-topics.sh--create --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep3 --partitions 5 --replication-factor 3/bin/kafka-topics.sh--create --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep4 --partitions 5 --replication-factor 4/bin/kafka-topics.sh--create --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep5 --partitions 5 --replication-factor 52、生成数据/bin/kafka-producer-perf-test.sh --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topics test_kafka_rep --messages 10000000  --message-size 4096 --batch-size 10000  --threads 1 --compression-codec 3  --hide-header3、消费数据./bin/kafka-consumer-perf-test.sh  --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_kafka_rep --messages 10000000 --fetch-size  1048576  --threads 5--num-fetch-threads 14、初步结论备份数越多,吞吐率越低。
  • 测试结果
  • 测试结论

数量过少会影响数据的可用性,太多则会白白浪费存储资源,一般建议在2~4为宜,我们设置为3个,既能保障数据的高可用,又避免了浪费过多的存储资源。

4.2.5 fetch-threads

  • 测试脚本:
./kafka-consumer-perf-test.sh --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_perf --fetch-size 1048576 --messages 1000000 --threads 5 --num-fetch-threads 1./kafka-consumer-perf-test.sh --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_perf --fetch-size 1048576 --messages 10000000 --threads 5 --num-fetch-threads 3./kafka-consumer-perf-test.sh --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_perf --fetch-size 1048576 --messages 50000000 --threads 5 --num-fetch-threads 5./kafka-consumer-perf-test.sh --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_perf --fetch-size 1048576 --messages 100000000 --threads 5  --num-fetch-threads 7./kafka-consumer-perf-test.sh --zookeeper 10.240.0.9:2181,10.240.0.10:2181,10.240.0.13:2181/kafka --topic test_perf --fetch-size 1048576 --messages 100000000 --threads 5  --num-fetch-threads 10
  • 测试结果

Threads|fetch-size|partition|replication|fetch-threads|MB/S|MsgNum/s --|--|--|--|--|--|--|-- 1|4194304| 1 | 1 | 1 | 117.86 | 179881 1|4194304| 1 | 1 | 3 | 137.56 | 209973 1|4194304| 1 | 1 | 6 | 127.63| 194810 1|4194304| 1 | 1 | 7 | 126.73 | 193434 1|4194304| 1 | 1 | 10 | 146.40 | 223458

  • 测试结论

在我们控制其他条件不变的情况下,我们更改fetch-thread的线程数,可以发现是随着线程数增多而消费速度加快,在fetch-threads=10时,最优为146.4m/s。

4.3 broker测试

涉及的参数众多,诸如以下:

  • default.replication.factor
  • num.replica.fetchers
  • auto.create.topics.enable
  • min.insync.replicas
  • unclean.leader.election.enable
  • broker.rack
  • log.flush.interval.messages
  • log.flush.interval.ms
  • unclean.leader.election.enable
  • min.insync.replicas
  • num.recovery.threads.per.data.dir

参数中很多都是我们调优方面需要的参数,对吞吐影响小的参数我们本次不进行测试,而产生影响的partition,replic以及IO的部分我们在producer和consumer中已得出结论,此处只进行和broker相关的系列参数的测试。影响参数具体如下:

  • 其中涉及到调优延时的num.replica.fetchers,如果发生ISR频繁进出的情况或follower无法追上leader的情况则适当增加该值,但通常不要超过CPU核数+1,在reblance中会对我们的吞吐产生间接影响;
  • 涉及持久性的诸如:default.replication.factor我们已测, auto.create.topics.enable=false此处我们不允许自动创建主题, min.insync.replicas建议设置为replication factor-1 这块我们需要保证ISR中的版本不掉, unclean.leader.election.enable= false 是否允许不具备ISR资格的replicas选举为leader,这个为了数据准确性设置为false, broker.rack如果有机架信息,则最好设置上,保证数据在多个rack间的分布性以达到高持久化,log.flush.interval.messages和log.flush.interval.ms如果是特别重要的topic并且TPS本身也不高,则推荐设置成比较低的值,比如1,此处刷盘对我们数据的持久化是有影响的,后续进行测试;
  • 涉及到高可用的如下:min.insync.replicas=1最少ISR中需要保留的broker数,num.recovery.threads.per.data.dirlog.dirs中配置的目录数,此三者在配置中我们来设置;

4.3.1 num.replica.fetchers

  • 测试方法:

   首先我们的参数是在broker的配置文件中设置,该参数设置的依据是如果发生ISR频繁进出的情况或follower无法追上leader的情况则适当增加该值,但通常不要超过CPU核数+1,所以我们从1开始测试到Num(core)+1;

  • 测试结果:
  • 测试结论:

测试fetchers对吞吐的影响时,可以发现产生波动但是整体变化不大,因此我们还是选取在32时最优。

4.3.2 num.io.threads

  • 测试方法:

   该参数也在broker的配置文件中来配,主要进行磁盘io操作,高峰期可能有些io等待,因此配置需要大些。配置线程数量建议为cpu核数2倍,最大不超过3倍,我们从core的整数倍测到3倍;

  • 测试结果:
  • 测试结论:

io-thread对吞吐的影响是随着线程数增多而逐步上升的,但不宜超过整体核数的3倍,因此我们还是选取在96时最优。

4.3.3 num.network.threads

  • 测试方法:

   该参数也在broker的配置文件中来配,主要处理网络io,读写缓冲区数据,基本没有io等待,配置线程数量从1开始,测到核数+1;

  • 测试结果:
  • 测试结论:

net-thread对吞吐的影响是随着线程数增多而逐步上升的,但不宜超过整体核数,因此我们还是选取在32时最优。

4.3.4 interval.messages

  • 测试脚本:

   该参数也在broker的配置文件中来配,为大幅提高producer写入吞吐量,需要定期批量写文件,此配置为满足多大进行flush;

  • 测试结果:
  • 测试结论:

该项参数在测试中发现对吞吐的影响不是很大,通过测试,我们可以发现在messages数为2w时,性能较好。

4.3.5 interval.ms

  • 测试脚本:

   该参数也在broker的配置文件中来配,为大幅提高producer写入吞吐量,需要定期批量写文件,此配置为满足多长时间进行flush;

  • 测试结果:
  • 测试结论:

该项参数在测试中发现对吞吐的影响不是特别巨大,通过测试,由于刷盘还会以消息数为依据,因而此处我们设置为1w ms每次刷盘一次。

5. 容灾测试

5.1 broker宕机

5.1.1 broker集群宕机

这个做过测试后,发现具体情况如下:

即使设置了acks=all,但是如果整个集群都连接不上了,也是不能避免消息丢失的(重发次数到了设定的值,或者发送请求超时了都会导致生产者丢弃该条消息,发送下一条消息)。重试次数增多、发送请求超时这个参数设置长点可以减少“丢失”(丢失只是相对于消费者角度来说的,实际上是生产者由于超时或者重发次数限制丢弃了一些消息)的消息数。重试次数增多,发送请求超时增加都意味着对发送失败的消息进行更长时间的重发。因此相对来说,被生产者丢弃的消息数会少些。

5.1.2 部分broker宕机

   集群有n份replication,那么一般来说,挂掉n-1 个节点都是没关系的。挂掉的broker对原来的消息收发几乎不产生任何影响。

5.1.2.1 对consumer的影响

broker挂掉之后,分区自然也要重新平衡,这时候会对消费产生什么影响,具体实验如下。

实验过程:

  • 生产者发送n条信息;
  • 生产者发送完毕后马上kill一个broker;
  • 观察消费者状态(应该是阻塞了)。此处消费者每次消费完一条记录休眠2秒,方便我们有时间KILL BROKER。此外设置poll为1毫秒,方便执行休眠;
  • 重启broker并观察消费者是否继续消费,消费的条数和正确性怎样;

消息条数为10:

  • 发送前状况:所有leader负载均衡,ISR集合包含2个server,没有处在同步中的分区。
  • 发送后,KILL掉broker后状态:分区进行重新平衡,leader重新选举为0,ISR集合只有一个SERVER,所有分区处于under replicated状态。
  • 看剩下的BROKER上也可以看到消费者的重新组织。

结论:此时consumer仍然成功消费到所有消息,无重复,无丢失。不同发送量:

5.1.2.2 对producer的影响

   生产者发送过程中(消费者保持正常消费),kill broker之后生产者可能会有如下的警告,但是因为设置了重发,重新发送后会发送到那个正常的broker上(retries可以设置)。观察消费者端的消费情况可以发现消费者稍微消费了几条数据后,稍微被阻塞了一会,但是很快又能回复消费正确拿到所有数据。

5.1.2.3 对分区leader和ISR的影响

  • 在正常运行时,所有leader都是平衡的,而且ISR集合也是包含所有SERVER,所有分区都不处于under replicated状态。
  • 当其中一个broker被kill后,其变化如下(PS:此时处于under replicated状态,这个在kafka manager上可以看),此时已经很快的选举好了新leader,然后尝试重启server,重启后可以发现ISR集合很快同步完成(此时在kafka manager上的Under Replicated状态又变为false),但是leader还没有平衡。leader的平衡有参数leader.imbalance.check.interval.seconds来控制,默认为300秒,因此需要等待5分钟才会leader重新平衡。
  • 5分钟后查看,一切恢复如初。

5.2 磁盘故障

  • 磁盘故障情况:当某个broker上的磁盘发生故障时,分区leader在该broker上的分区都无法进行访问,broker server进程被阻塞。如果磁盘上的数据能及时恢复,并且磁盘重新进行工作的话,出现磁盘故障的broker就能够重新恢复服务。而在磁盘故障没修复之前,其实整个kafka集群是不可用的。因此对可用性要求比较高的场景下,如果某个broker由于磁盘故障而不能服务,可以考虑尽快下线该broker,触发分区复制,确保整个集群可用。
  • 磁盘故障恢复方法:首先第一件事还是下线出问题的broker,确保整个集群可用。然后尽快修复磁盘上的数据,然后重启broker。如果磁盘上的数据没法恢复,也没有关系。可以尽快替换健康的磁盘,然后重启broker,这样数据还是可以通过复制恢复过来的。

5.3 恢复能力

这个测试主要测试kafka进行数据恢复的能力。当kafka集群上有大量历史数据时,如果其中一个broker挂了,需要多少时间来完成恢复,同时对生产和消费会产生神马影响,是我们主要关心的内容。由于我们采用3个节点,并且admin.benchmark这个topic只有2份副本,也就是意味着只能容忍1个broker故障。如果超过1个broker故障,就会影响消息收发,需要尽快恢复broker。

实验过程:

  • 按照第5节的方法进行大量消息的收发(先在上面保留2亿条记录,然后按照原来3个生产者9个消费者的方式启动整个收发流程)
  • 在收发过程中下线broker1,观察对消息收发的影响(延迟变化、是否发生错误或者异常)
  • 过几小时后重新启动broker1,查看恢复的时长和对生产消费的影响。
  • 再次关掉broker1并且清除broker1上面所有的日志记录,查看集群需要多久时间恢复约10亿条记录。
  • PS:我们下线broker1的时间为14:48分,这个对照图来分析的时候请留意。

实验结果:

  • 下线broker1之后,消费者没有报错,生产者开始刷出大量报错,约1分钟之后,所有生产者均开始重新恢复发送。在下线broker1之后,该broker上面的leader分区无法访问。这时候需要重新选取分区,然后到新的broker上去获取分区数据。而且还要触发复制。整个异常过程中,只有kafka集群的生产者会受影响,并且在较短的时间内自动恢复。
  • 下线broker1之后,broker1的网卡仍然占用着比较大的网络带宽,主要是复制分区数据导致。
  • 下线broker1之后,生产者的发送吞吐量降低、平均延迟增加、峰值时的延迟也增加。
  • 下线broker1之后的10分钟的时间内,消费者的消费速率会有显著下降,应该是要等待分区选举leader吧。不过总体上,消费者受影响仍然较小,总体上仍然是以比较正常的速率进行消费(平均有3W+条消息每秒的消费速率)。下图是broker2的一个消费者的消费统计信息,可供参考。
  • broker1下线后整个集群仍然正常提供服务。过几小时后重新恢复broker1,由于其节点上原来的数据仍然是保留的,所以整个集群马上可以恢复。
  • broker1如果下线后删除上面10亿条记录再上线,发现数据恢复需要耗时较久。

5.4 恢复能力

kafka作为分布式的消息系统,在集群可用性上还是做得比较完善的。在副本数充足的情况下发生节点故障,只会对生产和消费的速率产生一些影响,总体系统仍然是可用的。

而针对突发的大量消息收发,kafka集群能非常稳定的工作。从实验结果我们也可以看到,即使使用万兆网卡,我们的生产和消费都快要跑满整块网卡的带宽。一般来说,只要网络带宽给力,kafka的吞吐性能绝对是够用了(前提是生产及发送者本身不是性能瓶颈)。

目前我们遇到的问题故障恢复问题还有就是在reblance过程中,特别是虚拟机在物理机上会不定时的产生leader和follower的切换,这在我们生产消费时其实是会存在数据的抖动,此时设置好重试次数以及exactly-once策略很重要。

6. 单台机器测试

前置条件,一台机器,主题也是单分区单副本。

6.1 ack=1&&compression=lz4

在这种情况下,kafka使用虚拟内存为65.90g,而实际占用内存2.7g,cpu使用率最高能达到305%,内存占用率达到4.9%;

从进程来看,cpu占用率在304%,内存使用率则在5.2%;

从吞吐来看,达到了233409条/s,167.34m/s;

IO方面,我们的写入量峰值在19.82w/s,数据量达到4495wkB/s;

6.2 ack=0&&compression=lz4

在这种情况下,kafka使用虚拟内存为65.91g,而实际占用内存3.0g,cpu使用率最高能达到323.9%,内存占用率达到5.5%;

从进程来看,cpu占用率在259%,内存使用率则在5.4%;

从吞吐来看,达到了350988 r/s,229.96m/s;

IO方面,我们的写入量峰值在19.82w/s,数据量达到4495wkB/s;

6.3 ack=0&&compression=none

在这种情况下,kafka使用虚拟内存为65.93g,而实际占用内存3.1g,cpu使用率最高能达到204.7%,内存占用率达到5.7%;

从进程来看,cpu占用率在253%,内存使用率则在5.6%;

从吞吐来看,达到了72673条/s,47.61m/s;

IO方面,我们的写入量峰值在19.82w/s,数据量达到4495wkB/s;

6.4 ack=1&&compression=none

在这种情况下,kafka使用虚拟内存为66.42g,而实际占用内存3.14g,cpu使用率最高能达到232.5%,内存占用率达到5.7%;

从进程来看,cpu占用率在181%,内存使用率则在5.7%;

从吞吐来看,达到了71676条/s,46.96m/s;

IO方面,我们的写入量峰值在19.82w/s,数据量达到4495wkB/s;

7.测试结论

通过目前对服务器磁盘,kafka集群broker,producer,consumer的吞吐及生产消费的性能测试:

  • producer方面,在主从同步选取1时性能和稳定性适中,压缩方面,我们选择lz4压缩方式,而批大小我们可以选择100w左右,并发保持在60,消息的大小建议在4k左右较好,分区数在3-5个,副本数为3个既可以保证性能也能维持高可用;
  • 而consumer的处理线程我们选择4个,抓取消息大小则设置在400w条左右,抓取线程设置为10个即可;
  • broker的参数方面,replica.fetcher设置为服务器core的个数时较好,io.threads 则设置为core个数的3倍,network.threads保持和core个数相等即可,interval.messages数设置为2w,interval.ms则设置为10000 ms;
  • 在5章的单机器测试中,我们lz4的最优的性能达到了350988 r/s,229.96m/s,是比之前134上测试的211779r/s,138.75m/s要告出一部分的,因为之前134和143云服务器实际是基于一台物理机,而149和146则是在一台机器上,可能因为之前3个分区3个副本的方式因为这点会有影响,所以可以看出来,独自的kafka服务器目前可以得到的最好吞吐量在35w/s左右,磁盘IO写在19.82w/s,CPU占用率最高在323%,内存使用率在6%左右。

本文分享自微信公众号 - 大数据技术与架构(import_bigdata)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-08-23

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券