5.1 性能调优全攻略
Producer调优
批量发送与延迟发送
通过调整batch.size和linger.ms参数提升吞吐量：
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 默认16KB
props.put(ProducerConfig.LINGER_MS_CONFIG, 10); // 等待10ms以积累更多消息
batch.size：批量发送的字节数，达到该大小或linger.ms超时即发送。linger.ms：消息在缓冲区的最大停留时间，即使未达到batch.size也会发送。
压缩算法选择
启用压缩可显著减少网络传输和磁盘存储开销：
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"); // 可选：gzip、snappy、lz4、zstd
- Snappy：压缩速度快，压缩比适中。
- LZ4：压缩比和速度平衡，推荐大多数场景。
- ZSTD：压缩比最高，但CPU开销较大。
Broker调优
内存与线程配置
调整Broker的网络和IO线程池大小：
# server.properties
num.network.threads=8 # 网络处理线程数，默认3
num.io.threads=16 # IO处理线程数，默认8
socket.send.buffer.bytes=102400 # 发送缓冲区大小，默认100KB
socket.receive.buffer.bytes=102400 # 接收缓冲区大小，默认100KB
磁盘与日志管理
优化日志存储和清理策略：
# 日志段滚动大小，默认1GB
log.segment.bytes=536870912
# 日志保留时间，默认7天
log.retention.hours=168
# 日志清理策略：delete（按时间删除）或compact（按key压缩）
log.cleanup.policy=delete
# 后台日志清理线程数
log.cleaner.threads=2
Consumer调优
并行消费与反序列化优化
增加Consumer实例数或使用多线程消费：
// 增加Consumer Group中的Consumer数量，实现分区级并行
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(props);
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(props);
consumer1.subscribe(Collections.singletonList("topic"));
consumer2.subscribe(Collections.singletonList("topic"));
// 或在单个Consumer中使用多线程处理消息
ExecutorService executor = Executors.newFixedThreadPool(10);
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
executor.submit(() -> process(record));
}
}
使用高效的序列化格式（如Protobuf替代JSON）：
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ProtobufSerializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ProtobufDeserializer.class.getName());
5.2 实战场景模拟
场景一：高并发日志采集（每秒10W+消息写入）
架构设计
- Topic配置：创建100个分区的Topic，利用多分区并行写入提升吞吐量。
- Producer配置：
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768); // 32KB批次 props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // 5ms延迟 props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"); props.put(ProducerConfig.ACKS_CONFIG, "1"); // 牺牲部分可靠性换取高吞吐量 - Broker配置：
num.partitions=100 # 默认分区数 log.flush.interval.messages=100000 # 每10W条消息刷盘一次 log.flush.interval.ms=10000 # 每10秒刷盘一次
性能测试
使用kafka-producer-perf-test.sh工具测试写入性能：
bin/kafka-producer-perf-test.sh --topic log-topic --num-records 10000000 \
--record-size 100 --throughput -1 --producer-props bootstrap.servers=localhost:9092
场景二：实时数据分析（电商实时大屏）
数据流设计
- 数据源：用户浏览、下单、支付等行为数据实时写入Kafka。
- 流处理：Kafka Streams计算实时指标（如UV、GMV、转化率）：
KStream<String, String> userEvents = builder.stream("user-events-topic");
KTable<Windowed<String>, Long> hourlyUV = userEvents
.selectKey((key, value) -> value.getUserId())
.groupByKey()
.windowedBy(TimeWindows.of(Duration.ofHours(1)))
.count(Materialized.as("hourly-uv-store"));
hourlyUV.toStream()
.map((windowedKey, count) -> new KeyValue<>(windowedKey.key(), count))
.to("hourly-uv-topic", Produced.with(Serdes.String(), Serdes.Long()));
- 结果存储：计算结果写入Redis，供前端大屏实时查询。
性能优化
- Kafka配置：
# 减少消息延迟 queued.max.requests=1000 replica.lag.time.max.ms=30000 - Kafka Streams配置：
config.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024); // 10MB缓存 config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); // 1秒提交一次
场景三：金融级数据一致性（事务消息实现分布式事务）
架构设计
- 订单服务：接收用户订单请求，发送订单创建消息到Kafka。
- 库存服务：消费订单消息，扣减库存，发送库存扣减结果。
- 支付服务：消费库存扣减结果，处理支付，发送支付结果。
事务消息实现
// 初始化事务
producer.initTransactions();
try {
producer.beginTransaction();
// 发送订单创建消息
producer.send(new ProducerRecord<>("order-topic", orderId, order));
// 执行本地事务（如更新订单状态）
orderService.updateOrderStatus(orderId, "PROCESSING");
// 提交事务
producer.commitTransaction();
} catch (Exception e) {
// 回滚事务
producer.abortTransaction();
}
幂等性保障
消费端通过唯一ID去重，确保同一消息只处理一次：
@KafkaListener(topics = "inventory-topic")
public void processInventory(InventoryMessage message) {
// 检查是否已处理过
if (inventoryService.isProcessed(message.getId())) {
return;
}
// 处理库存扣减
inventoryService.decreaseStock(message.getProductId(), message.getQuantity());
// 标记为已处理
inventoryService.markAsProcessed(message.getId());
}
