Kafka性能測試實例
在分布式實時數(shù)據(jù)流場景下，隨著數(shù)據(jù)量的增長，對Kafka集群的性能和穩(wěn)定性的要求也很高。本篇博客將從生產(chǎn)者和消費(fèi)者兩方面來做性能測試，針對具體的業(yè)務(wù)和數(shù)據(jù)量，來調(diào)優(yōu)Kafka集群。
本次測試的環(huán)境信息由三臺物理機(jī)組成，具體信息如下所示：
Kafka系統(tǒng)提供了測試工具kafka-producer-perf-test.sh和kafka-consumer-perf-test.sh，通過該工具可以對生產(chǎn)者性能和消費(fèi)者性能進(jìn)行測試，獲取一組最佳的參數(shù)值，進(jìn)而提升生產(chǎn)者的發(fā)送效率和消費(fèi)者的讀取效率。這里如果需要實現(xiàn)帶有線程參數(shù)功能的工具，可以修改工具源代碼，新建一個kafka-producer-perf-test-0.8.sh腳本，實現(xiàn)內(nèi)容如下：
# 使用老版本的ProducerPerformance工具類 exec $(dirname $0)/kafka-run-class.sh kafka.tools.ProducerPerformance "$@"
生產(chǎn)者測試，分別從線程數(shù)、分區(qū)數(shù)、副本數(shù)、Broker數(shù)、同步與異步模式、批處理大小、消息長度大小、數(shù)據(jù)壓縮等維度來進(jìn)行。
創(chuàng)建一個擁有6個分區(qū)、1個副本的Topic，設(shè)置不同的線程數(shù)并發(fā)送相同的數(shù)據(jù)量，查看性能變化。測試腳本如下：
# 創(chuàng)建主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_producer_perf --partitions 6 --replication-factor 1 # 設(shè)置1個線程數(shù) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf --threads 1 --broker-list dn1:9092, dn2:9092, dn3:9092 # 設(shè)置10個線程數(shù) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf --threads 10 --broker-list dn1:9092, dn2:9092, dn3:9092 # 設(shè)置20個線程數(shù) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf --threads 20 --broker-list dn1:9092, dn2:9092, dn3:9092 # 設(shè)置25個線程數(shù) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf --threads 25 --broker-list dn1:9092, dn2:9092, dn3:9092 # 設(shè)置30個線程數(shù) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf --threads 30 --broker-list dn1:9092, dn2:9092, dn3:9092
向一個擁有6個分區(qū)、1個副本的Topic中，發(fā)送500萬條消息記錄時，隨著線程數(shù)的增加，每秒發(fā)送的消息記錄會逐漸增加。在線程數(shù)為25時，每秒發(fā)送的消息記錄達(dá)到最佳值，隨后再增加線程數(shù)，每秒發(fā)送的消息記錄數(shù)反而會減少。
（1）新建一個擁有12個分區(qū)、1個副本的主題；
（2）新建一個擁有24個分區(qū)、1個副本的主題；
（3）向擁有12個分區(qū)、1個副本的主題中發(fā)送相同數(shù)量的消息記錄，查看性能變化；
（4）向擁有24個分區(qū)、1個副本的主題中發(fā)送相同數(shù)量的消息記錄，查看性能變化。
執(zhí)行命令如下：
# 創(chuàng)建一個擁有12個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_producer_perf_p12 --partitions 12 --replication-factor 1 # 創(chuàng)建一個擁有24個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_producer_perf_p24 --partitions 24 --replication-factor 1 # 用一個線程發(fā)送數(shù)據(jù)到擁有12個分區(qū)的主題中 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_p12 --threads 1 --broker-list dn1:9092, dn2:9092, dn3:9092 # 用一個線程發(fā)送數(shù)據(jù)到擁有24個分區(qū)的主題中 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_p24 --threads 1 --broker-list dn1:9092, dn2:9092, dn3:9092
從測試結(jié)果來看，分區(qū)數(shù)越多，單線程生產(chǎn)者的吞吐量越小。
（1）創(chuàng)建一個擁有兩個副本、6個分區(qū)的主題；
（2）創(chuàng)建一個擁有3個副本、6個分區(qū)的主題；
（3）向擁有兩個副本、6個分區(qū)的主題中發(fā)送相同數(shù)量的消息記錄，查看性能變化；
（4）向擁有3個副本、6個分區(qū)的主題中發(fā)送相同數(shù)量的消息記錄，查看性能變化；
執(zhí)行命令如下：
# 創(chuàng)建一個擁有兩個副本、6個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_producer_perf_r2 --partitions 6 --replication-factor 2 # 創(chuàng)建一個擁有3個副本、6個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_producer_perf_r3 --partitions 6 --replication-factor 3 # 用3個線程發(fā)送數(shù)據(jù)到擁有兩個副本的主題中 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_r2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 # 用3個線程發(fā)送數(shù)據(jù)到擁有3個副本的主題中 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_r3 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092
從測試結(jié)果來看，副本數(shù)越多，吞吐量越小。
通過增加Broker節(jié)點(diǎn)數(shù)量來查看性能變化，腳本如下：
# Kafka節(jié)點(diǎn)數(shù)為4個時，異步發(fā)送消息記錄 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_b3 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092, dn4:9092 --batch-size 3000 --request-timeout-ms 100000
從測試結(jié)果來看，增加Kafka Broker數(shù)量，吞吐量會增加。
分別使用同步和異步模式發(fā)送相同數(shù)量的消息記錄，查看性能變化。執(zhí)行腳本如下：
# 創(chuàng)建一個有用3個副本、6個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_producer_perf_s2 --partitions 6 --replication-factor 3 # 使用同步模式發(fā)送消息數(shù)據(jù) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --sync # 使用異步模式發(fā)送消息記錄 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092
從測試結(jié)果來看，使用異步模式發(fā)送消息數(shù)據(jù)，比使用同步模式發(fā)送消息數(shù)據(jù)，吞吐量是同步模式的3倍左右。
使用異步模式發(fā)送相同數(shù)量的消息數(shù)據(jù)，改變批處理量的大小，查看性能變化，執(zhí)行腳本如下：
# 以批處理模式發(fā)送，大小為1000條 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 1000 --request-timeout-ms 100000 # 以批處理模式發(fā)送，大小為3000條 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 3000 --request-timeout-ms 100000 # 以批處理模式發(fā)送，大小為5000條 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 5000 --request-timeout-ms 100000 # 以批處理模式發(fā)送，大小為7000條 [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 7000 --request-timeout-ms 100000
3.6.2 結(jié)論
從測試的結(jié)果來看，發(fā)送的消息隨著批處理大小增加而增加。當(dāng)批處理大小增加到3000~5000時，吞吐量達(dá)到最佳值。而后再增加批處理大小，吞吐量的性能會下降。
改變消息的長度大小，查看性能變化，執(zhí)行腳本如下：
# 發(fā)送消息，長度為100字節(jié) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 3000 --request-timeout-ms 100000 --message-size 100 # 發(fā)送消息，長度為200字節(jié) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 3000 --request-timeout-ms 100000 --message-size 200 # 發(fā)送消息，長度為500字節(jié) [hadoop@dn1 ~]$ kafka-producer-perf-test-0.8.sh --messages 5000000 --topics test_producer_perf_s2 --threads 3 --broker-list dn1:9092, dn2:9092, dn3:9092 --batch-size 3000 --request-timeout-ms 100000 --message-size 500
從測試結(jié)果來看，隨著消息長度的增加，每秒所能發(fā)送的消息數(shù)量逐漸減少（nMsg/sec）。但是，每秒發(fā)送的消息的總大?。∕B/sec），會隨著消息長度的增加而增加。
消費(fèi)者測試，可以從線程數(shù)、分區(qū)數(shù)、副本數(shù)等維度來進(jìn)行測試。
創(chuàng)建一個擁有6個分區(qū)、1個備份的Topic，用不同的線程數(shù)讀取相同的數(shù)據(jù)量，查看性能變化。測試腳本如下：
# 創(chuàng)建主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_consumer_perf --partitions 6 --replication-factor 1 # 設(shè)置1個線程數(shù) [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh –zookeeper dn1:2181,dn2:2181,dn3:2181 --messages 5000000 --topic test_consumer_perf --group g1 --threads 1 # 設(shè)置3個線程數(shù) [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh –zookeeper dn1:2181,dn2:2181,dn3:2181 --messages 5000000 --topic test_consumer_perf --group g2 --threads 3 # 設(shè)置6個線程數(shù) [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh –zookeeper dn1:2181,dn2:2181,dn3:2181 --messages 5000000 --topic test_consumer_perf --group g3 --threads 6
隨著線程數(shù)的增加，每秒讀取的消息記錄會逐漸增加。在線程數(shù)與消費(fèi)主題的分區(qū)相等時，吞吐量達(dá)到最佳值。隨后，再增加線程數(shù)，新增的線程數(shù)將會處于空閑狀態(tài)，對提升消費(fèi)者程序的吞吐量沒有幫助。
新建一個Topic，改變它的分區(qū)數(shù)，讀取相同數(shù)量的消息記錄，查看性能變化，執(zhí)行腳本如下：
# 創(chuàng)建一個擁有12個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_consumer_perf_p12 --partitions 12 --replication-factor 1 # 創(chuàng)建一個擁有24個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 --topic test_consumer_perf_p24 --partitions 24 --replication-factor 1 # 用一個線程讀取數(shù)據(jù)到擁有12個分區(qū)的主題中 [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh –zookeeper dn1:2181,dn2:2181,dn3:2181 --messages 5000000 –topic test_consumer_perf_p12_--group g2 --threads 1 # 用一個線程讀取數(shù)據(jù)到擁有12個分區(qū)的主題中 [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh –zookeeper dn1:2181,dn2:2181,dn3:2181 --messages 5000000 –topic test_consumer_perf_p24_--group g3 --threads 1
當(dāng)分區(qū)數(shù)增加時，如果線程數(shù)保持不變，則消費(fèi)者程序的吞吐量性能會下降。
新建Topic，改變Topic的副本數(shù)，讀取相同數(shù)量的消息記錄，查看性能變化，執(zhí)行腳本如下：
# 創(chuàng)建一個有用兩個副本、6個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 –topic test_consumer_perf_r2 --partitions 6 --replication-factor 2 # 創(chuàng)建一個有3個副本、6個分區(qū)的主題 [hadoop@dn1 ~]$ kafka-topics.sh --create --zookeeper dn1:2181, dn2:2181, dn3:2181 –topic test_consumer_perf_r3 --partitions 6 --replication-factor 3 # 用3個線程讀取數(shù)據(jù)到擁有兩個副本的主題中 [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh –zookeeper dn1:2181 ,dn2:2181,dn3:2181 --messages 5000000 –topic test_consumer_perf_r2_--group g2 --threads 3 # 用3個線程讀取數(shù)據(jù)到擁有3個副本的主題中 [hadoop@dn1 ~]$ kafka-consumer-perf-test.sh --zookeeper dn1:2181 ,dn2:2181,dn3:2181 --messages 5000000 –topic test_consumer_perf_r3_--group g3 --threads 3
4.3.2 結(jié)論
副本數(shù)對消費(fèi)者程序的吞吐量影響較小，消費(fèi)者程序是從Topic的每個分區(qū)的Leader上讀取數(shù)據(jù)的，而與副本數(shù)無關(guān)。
Kafka性能測試步驟并不復(fù)雜，大家可以根據(jù)實際的測試環(huán)境、數(shù)據(jù)量，通過對生產(chǎn)者和消費(fèi)者不同維度的測試，來獲取一組最佳的調(diào)優(yōu)參數(shù)值。
