Kafka 性能调优与容量规划：从生产者到 Broker 的全链路压测指南

       吞吐（Throughput）
       /            \
      /              \
   延迟             持久化
  (Latency)        (Durability)

CPU 高 → 压缩/序列化/协议开销（加密、压缩 CPU 密集）
磁盘 IO 高 → 写入跟不上（换 SSD / 降副本 / 缩消息）
网络高 → 跨机房/副本同步吃带宽
内存不足 → 页缓存小、GC 压力大

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);        // 批量大小 32KB
props.put(ProducerConfig.LINGER_MS_CONFIG, 20);            // 攒 20ms 再发
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");  // 压缩类型
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);  // 发送缓冲 32MB
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1048576);

// 低延迟（订单支付、实时告警）：linger 小、批小
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");

// 高吞吐（埋点、日志管道）：linger 大、压缩开
props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 131072);  // 128KB
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");

props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);        // 攒满才返回
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1048576);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

① 分区数：Topic 分区数决定最大并行度
   → 消费者实例 ≤ 分区数
② 每实例多线程：单实例内线程池处理（注意偏移提交与顺序）
③ 批量处理：一次 poll 批量入库，避免逐条 RPC

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    // 批量插入，替代逐条处理
    bulkInsert(records);          // 攒一批 JDBC batch insert
    consumer.commitSync();        // 成功后一次性提交
}

# broker 关键配置
num.network.threads=8
num.io.threads=8           # 处理请求的 IO 线程
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
log.flush.interval.messages=10000
log.segment.bytes=1073741824   # segment 1GB
log.retention.hours=168
log.dirs=/data/kafka1,/data/kafka2   # 多盘分散 IO

分区数过低 → 消费并行度不足、单分区热点
分区数过高 → 元数据/文件句柄/协调开销上升
经验公式：
  目标吞吐 / 单分区吞吐(≈10-20MB/s) ≈ 所需分区数
  再考虑：峰值 / 3（副本放大）留余量

JSON：可读、慢、体积大
Avro/Protobuf：快、小、需 Schema Registry
String/ByteArray：最简，适合二进制载荷

小消息 + 大 batch + zstd → 压缩率最高
大消息（>100KB）→ 压缩收益递减，评估是否值得

场景：单 Topic 峰值 20万 msg/s，单条 1KB
  单分区吞吐 ≈ 10-20 MB/s ≈ 1万-2万 msg/s
  所需分区数 ≈ 20万 / 1.5万 ≈ 14 分区（留余量取 18-24）

写放大：3 副本 → 磁盘实际写入 = 20万 × 1KB × 3 ≈ 600MB/s
带宽：600MB/s > 千兆(125MB/s) → 需万兆或压缩（1KB→200B → 120MB/s ✅）

retention 7 天，20万 msg/s × 1KB × 3 副本
  = 200MB/s × 86400s × 7天 ≈ 120TB（原始未压缩）
  压缩 5× → ≈ 24TB
规划时按「压缩后 × 1.5 余量」备盘

# 生产者压测：发送 100 万条 1KB 消息
bin/kafka-producer-perf-test.sh \
  --topic perf-test --num-records 1000000 \
  --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers=localhost:9092 \
  acks=1 linger.ms=20 compression.type=lz4

# 消费者压测
bin/kafka-consumer-perf-test.sh \
  --topic perf-test --messages 1000000 \
  --threads 3 --broker-list localhost:9092

① 单测变量：一次只改一个参数，记录吞吐/延迟
② 梯度加压：记录 P50/P95/P99 延迟曲线，找拐点
③ 端到端验证：不只测单点，测「生产→消费」全链路
④ 持续观察：压测时看 CPU/磁盘/网络/GC 四象限