Kafka性能测试

测试背景

Apache Kafka是一种高吞吐、可扩展的分布式消息队列服务，它最初由LinkedIn公司开发，最后发展为Apache基金会的一个项目。目前kafka已经广泛应用于大数据分析，消息处理等环境，官方文档介绍kafka为提高吞吐率做了很多设计，但是其性能究竟如何呢？本文对kafka在不同参数下的性能进行测试。

测试目标

测试kafka 0.8n的性能(Producer/Consumer性能)。当消息大小、批处理大小、压缩等参数变化时对吞吐率的影响。

测试环境

软件版本：kafka 0.8.1.1
硬件环境：3台多磁盘服务组成的kafka集群。各服务器CPU E5645，内存47G，12快SAS盘，配置千兆网卡，配置如下：

测试方法

使用kafka官方提供的kafa-perf工具做性能测试，在测试时使用ganglia，kafka Web Console来记录服务情况。

测试步骤

一、测试环境准备
1、测试工具kafka-perf编译
kafka官方提供的二进制版本，并不包括性能测试的jar包，会报错找不到ProducerPerformance，要自己重新编译，编译方法如下：

git clone https://git-wip-us.apache.org/repos/asf/kafka.git kafka.git
cd kafka.git/
git checkout -b 0.8.1 origin/0.8.1
vim build.gradle   #编辑配置
./gradlew -PscalaVersion=2.10.4 perf:jar #生成2.10.4的kafka-perf的jar包，复制到libs目录下
cp perf/build/libs/kafka-perf_2.10-0.8.1.1-SNAPSHOT.jar /usr/local/kafka/libs/

2、启动kafka

cd /usr/local/kafka
vim config/server.properties  #内容见下图
./bin/kafka-server-start.sh config/server.properties &

kafka-producer-perf-test.sh中参数说明：

messages 生产者发送走的消息数量
message-size 每条消息的大小
batch-size 每次批量发送消息的数量
topics 生产者发送的topic
threads 生产者
broker-list 安装kafka服务的机器ip:porta列表
producer-num-retries 一个消息失败发送重试次数
request-timeouts-ms 一个消息请求发送超时时间

bin/kafka-consumer-perf-test.sh中参数说明：

zookeeper  zk配置
messages 消费者消费消息的总数量
topic 消费者需要消费的topic
threads 消费者使用几个线程同时消费
group 消费者组名称
socket-buffer-sizes socket缓存大小
fetch-size 每次想kafka broker请求消费消息大小
consumer.timeout.ms 消费者去kafka broker拿一条消息的超时时间

二、测试生产者吞吐率
此项只测试producer在不同的batch-zie，patition等参数下的吞吐率，也就是数据只被及计划，没有consumer读取数据消费情况。
生成Topic：
生成不同复制因子，partition的topic

bin/kafka-topics.sh --zookeepr 10.x.x.x:2181/kafka/k1001 --create --topic test-pati1-rep1 --partitions 1 --replication-factor 1
bin/kafka-topics.sh --zookeepr 10.x.x.x:2181/kafka/k1001 --create --topic test-pati1-rep2 --partitions 1 --replication-factor 2

bin/kafka-topics.sh --zookeepr 10.x.x.x:2181/kafka/k1001 --create --topic test-pati10-rep1 --partitions 10 --replication-factor 1
bin/kafka-topics.sh --zookeepr 10.x.x.x:2181/kafka/k1001 --create --topic test-pati10-rep2 --partitions 10 --replication-factor 2

bin/kafka-topics.sh --zookeepr 10.x.x.x:2181/kafka/k1001 --create --topic test-pati100-rep1 --partitions 100 --replication-factor 1
bin/kafka-topics.sh --zookeepr 10.x.x.x:2181/kafka/k1001 --create --topic test-pati100-rep2 --partitions 100 --replication-factor 2

测试producer吞吐率
调整batch-size，thread，topic，压缩等参数测试producer吞吐率。
示例：

a)批处理为1，线程数为1，partition为1，复制因子为1
bin/kafka-producer-perf-test.sh --messages 2000000 --message-size 512 --batch-size 1 --topic test-pati1-rep1 
--partitions 1 --threads 1 --broker-list host1:9092,host2:9092,host3:9092

b)批处理为10，线程数为1，partition为1，复制因子为2
bin/kafka-producer-perf-test.sh --messages 2000000 --message-size 512 --batch-size 10 --topic test-pati1-rep2 
--partitions 1 --threads 1 --broker-list host1:9092,host2:9092,host3:9092

c)批处理为100，线程数为10，partition为10，复制因子为2,不压缩，sync
bin/kafka-producer-perf-test.sh --messages 2000000 --message-size 512 --batch-size 100 --topic test-pati10-rep2 --partitions 10 --threads 10 --compression-codec 0 --sync 1 --broker-list host1:9092,host2:9092,host3:9092

d)批处理为100，线程数为10，partition为10，复制因子为2,gzip压缩，sync
bin/kafka-producer-perf-test.sh --messages 2000000 --message-size 512 --batch-size 100 --topic test-pati10-rep2 --partitions 10 --threads 10 --compression-codec 1 --sync 1 --broker-list host1:9092,host2:9092,host3:9092

说明：消息大小统一使用和业务场景中日志大小相近的512Bype，消息数为50w或200w条。
 
三、测试消费者吞吐率
测试consumer吞吐率
调整批处理数，线程数，partition数，复制因子，压缩等进行测试。
示例：

a)批处理为10，线程数为10，partition为1，复制因子为1
./bin/kafka-consumer-perf-test.sh --messages 500000 --batch-size 10 --topic test-pai1-rep2 --partitions 1 --threads 10 --zookeeper zkhost:2181/kafka/k1001

b)批处理为10，线程数为10，partition为10，复制因子为2
./bin/kafka-consumer-perf-test.sh --messages 500000 --batch-size 10 --topic test-pai1-rep2 --partitions 10 --threads 10 --zookeeper zkhost:2181/kafka/k1001

c)批处理为100，线程数为10，partition为1，复制因子为1,不压缩
./bin/kafka-consumer-perf-test.sh --messages 500000 --batch-size 100 --topic test-pai100-rep1 --partitions 1 --threads 10 --compression-codec 0 --zookeeper zkhost:2181/kafka/k1001

d)批处理为100，线程数为10，partition为1，复制因子为1,Snappy压缩
./bin/kafka-consumer-perf-test.sh --messages 500000 --batch-size 100 --topic test-pai100-rep1 --partitions 1 --threads 10 --compression-codec 2 --zookeeper zkhost:2181/kafka/k1001

测试结果及分析

1、生产者测试结果及分析
调整线程数，批处理数，复制因子等参考，对producer吞吐率进行测试。在测试时消息大小为512Byte，消息数为200w，结果如下：

调整sync模式，压缩方式得到吞吐率数据如表3.在本次测试中msg=512Byte，message=2000000，Partition=10，batch_zie=100

 
结果分析：
1）kafka在批处理，多线程，不适用同步复制的情况下，吞吐率是比较高的，可以达80MB/s,消息数达17w条/s以上。
2）使用批处理或多线程对提升生产者吞吐率效果明显。

3）复制因子会对吞吐率产生较明显影响
   使用同步复制时，复制因子会对吞吐率产生较明显的影响。复制因子为2比因子为1（即无复制）时，吞吐率下降40%左右。

4）使用sync方式，性能有明显下降。
   使用Sync方式producer吞吐率会有明显下降，表3中async方式最大吞吐率由82.0MB/s，而使用sync方式时吞吐率只有13.33MB/s.
 

5)压缩与吞吐率
  见图3，粉笔使用Gzip及Snappy方式压缩，吞吐率反而有下降，原因待分析。而Snappy方式吞吐率高于gzip方式。

6）分区数与吞吐率
   分区数增加生产者吞吐率反而有所下降 
 
2、消费者结果及分析
调整批处理数，分区数，复制因子等参数，对consumer吞吐率进行测试。在测试时消息大小为512Byte，消息数为200结果如下：  
 

调整压缩方式，分区数，批处理数等，测试参数变化时consumer的吞吐率。测试的复制因子为1。

结果分析：1）kafka consumer吞吐率在parition，threads较大的情况下，在测试场景下，最大吞吐率达到了123MB/s，消息数为25w条/s

2）复制因子，影响较小。replication factor并不会影响consumer的吞吐率测试，运维consumer智慧从每个partition的leader读数据，而与replication factor无关。同样，consumer吞吐率也与同步复制还是异步复制无关。

3)线程数和partition与吞吐率关系

图5为msg_size=512，batch_zie=100时的测试数据备份。可以看到当分区数较大时，如partion为100时，增加thread数可显著提升consumer的吞吐率。Thread10较thread1提升了10倍左右，而thread为100时叫thread为1提升了近20倍，达到120MB/s.

但要注意在分区较大时线程数不改大于分区数，否则会出现No broker partitions consumed by consumer，对提升吞吐率也没有帮助。图5中partion为10时，thread_10,thread_100吞吐率相近都为35MB/s左右。
4）批处理数对吞吐率影响
图表5中可以看出改变批处理数对吞吐率影响不大

5）压缩与吞吐率