螞蟻面試：Kafka 如何做壓測？如何保證系統穩定？
作者：甜獲師兄
Kafka是大數據領域應用非常廣泛的消息中間件，今天我們來介紹Kafka壓測方案，來確認Kafka集群的各類指標。
Kafka是大數據領域應用非常廣泛的消息中間件，如何確定Kafka集群的最大吞吐量和延遲呢？又如何保證Kafka集群的穩定呢？今天我們來介紹Kafka壓測方案，來確認Kafka集群的各類指標。
一、Kafka自帶性能測試工具
Kafka提供了內置的性能測試工具，可以用于生產者和消費者的基準測試：
- 生產者性能測試工具：kafka-producer-perf-test.sh
- 消費者性能測試工具：kafka-consumer-perf-test.sh
第三方壓測工具JMeter：
- 可以使用JMeter的Kafka插件進行壓測Tsung
- 支持Kafka協議的分布式壓測工具Gatling
- 可以通過Kafka插件進行壓測
二、壓測場景設計
1. 生產者性能測試
測試不同消息大小、批處理設置和壓縮算法對生產者性能的影響：
# 測試100字節消息，無壓縮
/opt/kafka/bin/kafka-producer-perf-test.sh \
--topic test-topic \
--num-records 10000000 \
--record-size 100 \
--throughput -1 \
--producer-props bootstrap.servers=broker1:9092,broker2:9092,broker3:9092 \
acks=1 \
batch.size=16384 \
linger.ms=0 \
compression.type=none
# 測試1KB消息，使用lz4壓縮
/opt/kafka/bin/kafka-producer-perf-test.sh \
--topic test-topic \
--num-records 10000000 \
--record-size 1024 \
--throughput -1 \
--producer-props bootstrap.servers=broker1:9092,broker2:9092,broker3:9092 \
acks=1 \
batch.size=65536 \
linger.ms=10 \
compression.type=lz42. 消費者性能測試
測試不同消費者組配置和分區數對消費性能的影響：
# 基本消費者性能測試
/opt/kafka/bin/kafka-consumer-perf-test.sh \
--bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
--topic test-topic \
--messages 10000000 \
--threads 1 \
--print-metrics
# 多線程消費者測試
/opt/kafka/bin/kafka-consumer-perf-test.sh \
--bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
--topic test-topic \
--messages 10000000 \
--threads 8 \
--print-metrics3. 端到端延遲測試
測量從生產到消費的端到端延遲：
# 創建一個具有多個分區的測試主題
/opt/kafka/bin/kafka-topics.sh \
--bootstrap-server broker1:9092 \
--create \
--topic latency-test \
--partitions 8 \
--replication-factor 3
# 使用自定義Java程序測量端到端延遲
// 生產者代碼示例
Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 10000; i++) {
long timestamp = System.currentTimeMillis();
ProducerRecord<String, String> record =
new ProducerRecord<>("latency-test", null, timestamp, "key-" + i, "value-" + timestamp);
producer.send(record);
Thread.sleep(100); // 每秒發送10條消息
}
producer.close();
// 消費者代碼示例
Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("group.id", "latency-test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("latency-test"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
long latency = System.currentTimeMillis() - record.timestamp();
System.out.printf("Offset = %d, Latency = %d ms%n", record.offset(), latency);
}
}4. 吞吐量與延遲權衡測試
測試不同配置下吞吐量與延遲的權衡關系：
# 高吞吐量配置測試
/opt/kafka/bin/kafka-producer-perf-test.sh \
--topic throughput-test \
--num-records 5000000 \
--record-size 1024 \
--throughput -1 \
--producer-props bootstrap.servers=broker1:9092,broker2:9092,broker3:9092 \
acks=1 \
batch.size=131072 \
linger.ms=50 \
compression.type=lz4 \
buffer.memory=67108864
# 低延遲配置測試
/opt/kafka/bin/kafka-producer-perf-test.sh \
--topic latency-test \
--num-records 1000000 \
--record-size 1024 \
--throughput -1 \
--producer-props bootstrap.servers=broker1:9092,broker2:9092,broker3:9092 \
acks=1 \
batch.size=8192 \
linger.ms=0 \
compression.type=none三、壓測指標分析
1. 生產者關鍵指標
- 吞吐量（Throughput）：每秒處理的消息數或字節數
- 延遲（Latency）：消息從發送到確認的時間
- CPU使用率：生產者進程的CPU使用情況
- 內存使用率：生產者進程的內存使用情況
- 批處理率：每批次的平均消息數
2. 消費者關鍵指標
- 吞吐量：每秒消費的消息數或字節數
- 延遲：消息從生產到消費的時間
- 消費者滯后（Consumer Lag）：消費者落后于生產者的消息數
- 處理時間：消費者處理每條消息的時間
- 提交率：偏移量提交的頻率和成功率
3. Broker關鍵指標
- 請求處理率：每秒處理的請求數
- 請求隊列大小：等待處理的請求數
- 網絡吞吐量：進出Broker的網絡流量
- 磁盤使用率：日志文件的增長速率
- GC暫停時間：垃圾收集對性能的影響
四、壓測結果解讀
1. 生產者性能分析
以下是一個典型的生產者性能測試結果示例：
100000 records sent, 25000.0 records/sec (24.41 MB/sec), 15.2 ms avg latency, 293.0 ms max latency.
200000 records sent, 26315.8 records/sec (25.67 MB/sec), 12.8 ms avg latency, 128.0 ms max latency.
300000 records sent, 27272.7 records/sec (26.61 MB/sec), 11.5 ms avg latency, 98.0 ms max latency.結果解讀：
- 吞吐量隨時間穩定在約26,000條記錄/秒（約25MB/秒）
- 平均延遲約為13毫秒，最大延遲為293毫秒
- 隨著測試進行，延遲趨于穩定，表明系統性能良好
2. 消費者性能分析
以下是一個典型的消費者性能測試結果示例：
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2023-05-01 10:00:00, 2023-05-01 10:01:00, 1024.00, 17.07, 1048576, 17476.27, 20, 60000, 17.07, 17476.27結果解讀：
- 消費速率為17.07MB/秒，約17,476條消息/秒
- 重平衡時間為20毫秒，表明消費者組協調效率高
- 獲取時間為60秒，與測試持續時間一致
3. 瓶頸識別與解決
常見的性能瓶頸及解決方案：
CPU瓶頸：
- 增加broker數量
- 優化消息壓縮算法
- 調整JVM參數
內存瓶頸：
- 增加堆內存大小
- 優化生產者/消費者客戶端緩沖區大小
- 減少不必要的對象創建
磁盤I/O瓶頸：
- 使用更快的存儲（如SSD）
- 增加數據目錄數量，分散I/O負載
- 優化日志段大小和刷盤策略
網絡瓶頸：
- 增加網絡帶寬
- 優化消息批處理大小
- 使用更高效的壓縮算法
責任編輯：趙寧寧
來源：
大數據技能圈
