5.1 性能调优全攻略
Producer调优
批量发送与延迟发送
通过调整batch.size和linger.ms参数提升吞吐量：
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 默认16KBprops.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.propertiesnum.network.threads=8 # 网络处理线程数，默认3num.io.threads=16 # IO处理线程数，默认8socket.send.buffer.bytes=102400 # 发送缓冲区大小，默认100KBsocket.receive.buffer.bytes=102400 # 接收缓冲区大小，默认100KB磁盘与日志管理
优化日志存储和清理策略：
# 日志段滚动大小，默认1GBlog.segment.bytes=536870912 # 日志保留时间，默认7天log.retention.hours=168 # 日志清理策略：delete（按时间删除）或compact（按key压缩）log.cleanup.policy=delete # 后台日志清理线程数log.cleaner.threads=2Consumer调优
并行消费与反序列化优化
增加Consumer实例数或使用多线程消费：
// 增加Consumer Group中的Consumer数量，实现分区级并行KafkaConsumer consumer1 = new KafkaConsumer<>(props);KafkaConsumer consumer2 = new KafkaConsumer<>(props);consumer1.subscribe(Collections.singletonList("topic"));consumer2.subscribe(Collections.singletonList("topic")); // 或在单个Consumer中使用多线程处理消息ExecutorService executor = Executors.newFixedThreadPool(10);while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord 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 userEvents = builder.stream("user-events-topic");KTable, 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=1000replica.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());}