4. 压测工具选型与部署：kafka-producer-perf-test、kafka-consumer-perf-test、JMeter插件、自定义压测脚本

在 Kafka 链路压力测试中，工具选型直接影响测试结果的可靠性与排查效率。本章围绕四种主流压测手段展开：Kafka 官方命令行工具、JMeter 插件扩展、以及自定义脚本方案。每种工具在吞吐量、延迟、资源开销、可编程性上各有侧重，需根据压测目标（单点性能验证 vs. 全链路异常复现）灵活组合。

4.1 工具选型对比矩阵

4.2 kafka-producer-perf-test 部署与关键参数

该工具位于 Kafka 二进制包 bin/ 目录下，无需额外安装。核心参数控制消息体大小、吞吐目标、确认机制与压缩类型。

# 典型生产者压测命令
bin/kafka-producer-perf-test \
  --topic perf-test-topic \
  --num-records 500000 \
  --record-size 1024 \
  --throughput 50000 \
  --producer-props bootstrap.servers=192.168.1.10:9092 \
                         acks=1 \
                         compression.type=lz4 \
                         linger.ms=5 \
                         batch.size=32768

4.3 kafka-consumer-perf-test 部署与关键参数

消费者压测工具同样位于 bin/ 目录，用于测量从拉取请求到消费完成的总端到端延迟。注意该工具默认使用 new consumer API，支持多线程消费。

# 典型消费者压测命令
bin/kafka-consumer-perf-test \
  --topic perf-test-topic \
  --messages 500000 \
  --group perf-consumer-group \
  --threads 3 \
  --bootstrap-server 192.168.1.10:9092 \
  --reporting-interval 1000

4.4 JMeter + Kafka Backend Listener 部署

JMeter 通过 kafka-backend-listener 插件将压测结果实时发送至 Kafka 主题，实现全链路监控。部署步骤分为插件安装与监听器配置。

4.4.1 插件安装

1. 从 JMeter Plugins Manager 安装 kafka-backend-listener（或手动下载 jar 放入 lib/ext/）。
2. 确保 lib/ 目录包含 kafka-clients 依赖（建议使用与集群匹配的版本）。
3. 重启 JMeter，在监听器列表中出现 Kafka Backend Listener。

4.4.2 监听器配置

# Kafka Backend Listener 关键配置项
kafka.topic = jmeter-metrics
kafka.bootstrap.servers = 192.168.1.10:9092
kafka.acks = 1
kafka.key.serializer = org.apache.kafka.common.serialization.StringSerializer
kafka.value.serializer = org.apache.kafka.common.serialization.StringSerializer
metrics.send.interval = 5000   # 每5秒推送一次聚合指标

4.5 自定义压测脚本（Python / Go）

当需要复现特定异常模式（如分区 leader 切换、ISR 收缩、网络分区）时，自定义脚本提供最大灵活性。以下给出 Go 语言生产者压测骨架，重点展示协程并发、上下文超时与指标采集。

package main

import (
    "context"
    "fmt"
    "time"
    "github.com/segmentio/kafka-go"
)

func main() {
    writer := &kafka.Writer{
        Addr:     kafka.TCP("192.168.1.10:9092"),
        Topic:    "perf-test-topic",
        Balancer: &kafka.LeastBytes{},
        BatchSize: 100,
        BatchTimeout: 10 * time.Millisecond,
        Async: true,
    }
    defer writer.Close()

    ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    defer cancel()

    msgCount := 500000
    latencyChan := make(chan time.Duration, msgCount)

    for i := 0; i < msgCount; i++ {
        go func(seq int) {
            start := time.Now()
            err := writer.WriteMessages(ctx, kafka.Message{
                Key:   []byte(fmt.Sprintf("key-%d", seq%100)),
                Value: []byte(fmt.Sprintf("{\"seq\":%d,\"ts\":%d}", seq, time.Now().UnixNano())),
            })
            if err != nil {
                fmt.Printf("write error: %v\n", err)
                return
            }
            latencyChan <- time.Since(start)
        }(i)
    }

    // 聚合延迟分位值（P50/P99/P999）
    // 此处省略聚合逻辑，实际部署时输出到 Prometheus 或日志
}

4.6 工具部署架构总览

4.7 部署检查清单

• 网络连通性： 压测机与所有 Kafka broker 之间双向 TCP 可达，防火墙放通 9092（或自定义端口）。
• 版本一致性： 压测工具使用的 kafka-clients 版本与集群 broker 版本大版本一致（如 2.8.x vs 3.x 协议有差异）。
• 资源预留： 压测机 CPU 核心数 ≥ 4，内存 ≥ 8 GB，避免 GC 抖动污染测试数据。
• 监控对齐： 压测开始前确保 broker 端 JMX 指标（如 kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec）已接入 Prometheus。
• 数据清理： 压测主题在测试前执行 kafka-delete-records 或设置 retention.ms=60000，避免历史数据堆积影响消费延迟。