一、一个参数引发的架构争议Kafka 生产者发送消息时acks参数决定了消息算不算发送成功acks 值含义可靠性吞吐量0生产者发完就不管了最低可能丢数据最高1Leader 写入成功即确认中等Leader 宕机可能丢较高all/-1所有 ISR 副本都写入成功最高最低争论永远围绕同一个问题acksall 到底慢多少值不值得有人说acksall 性能差 3 倍有人说几乎没影响。事实是——取决于你怎么测。本文用 6 组实验给你真实数据。二、acks 机制源码分析2.1 生产者发送链路2.2 Broker 端处理逻辑在 Broker 端消息写入的核心逻辑在ReplicaManager.appendRecords()中// Kafka ReplicaManager 源码简化版展示 acks 逻辑 def appendRecords( timeout: Long, requiredAcks: Short, // 就是 acks 参数 internalTopicsAllowed: Boolean, entriesPerPartition: Map[TopicPartition, MemoryRecords], responseCallback: Map[TopicPartition, PartitionResponse] Unit ): Unit { // 1. 参数校验acks 必须是 0, 1, 或 -1 if (requiredAcks ! 0 requiredAcks ! 1 requiredAcks ! -1) throw new InvalidRequiredAcksException(...) // 2. 写入 Leader 本地日志 val localProduceResults appendToLocalLog( internalTopicsAllowed, entriesPerPartition, requiredAcks ) // 3. 根据 acks 值决定响应策略 requiredAcks match { case 0 // acks0不需要等待任何确认立即回调 responseCallback(localProduceResults.mapValues(_.toPartitionResponse)) case 1 // acks1Leader 写入完成即可回调 // 如果 Leader 写入失败返回错误否则直接响应 responseCallback(localProduceResults.mapValues(_.toPartitionResponse)) case -1 // acksall需要等待所有 ISR 副本同步完成 // 创建 DelayedProduce挂起请求等待 follower 拉取 val produceMetadata ProduceMetadata(requiredAcks, localProduceResults) val delayedProduce new DelayedProduce( timeout, produceMetadata, this, responseCallback, localProduceResults ) // 放入 DelayedOperationPurgatory延迟操作队列 delayedProducePurgatory.tryCompleteElseWatch(delayedProduce) } }2.3 acksall 的延迟等待机制acksall时消息不是写完 Leader 就返回而是放入DelayedProducePurgatory等待所有 ISR 副本同步// DelayedProduce.tryComplete() 核心逻辑简化 override def tryComplete(): Boolean { produceMetadata.produceStatus.foreach { case (topicPartition, status) // 检查每个分区的副本同步状态 val partition replicaManager.getPartition(topicPartition) val leaderHWIncremented partition.getReplica(leaderReplicaId) match { case Some(leaderReplica) // 获取 Leader 的高水位HW val leaderHW leaderReplica.highWatermark // 检查所有 ISR 副本是否都拉取到了这条消息 status.requiredOffset match { case Some(requiredOffset) // 关键只有当 HW requiredOffset 时才算所有副本同步完成 if (leaderHW requiredOffset) { true // 同步完成 } else { return false // 还有副本没同步完继续等待 } case None // Leader 写入失败不需要等待 true } case None true } } // 所有分区都满足条件完成延迟操作 forceComplete() }关键点acksall的延迟取决于follower 副本拉取 Leader 数据的速度。如果 follower 落后很多延迟就会很高。2.4 ISR 与 acksall 的关系如果 ISR 中某个副本长时间不拉取acksall就会一直等待直到超时request.timeout.ms。三、压测环境与方案设计3.1 集群配置组件配置Broker 数量3 节点服务器规格8C 16G × 3SSD 500GBKafka 版本3.6.0Topic 配置6 分区 × 3 副本min.insync.replicas2消息大小1KB典型业务消息消息格式JSON3.2 生产者配置// 压测用 Producer 配置 Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker1:9092,broker2:9092,broker3:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // 核心变量 props.put(ProducerConfig.ACKS_CONFIG, acks); // 0 / 1 / all // 固定参数 props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 16KB batch props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // 最多等 5ms 凑 batch props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, lz4); // LZ4 压缩 props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB 缓冲 props.put(ProducerConfig.RETRIES_CONFIG, 3); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000); // 30s 超时3.3 压测场景场景编号场景名称说明S1单分区串行1 分区1 个 Producer 线程测试单线程极限S2多分区并行6 分区6 个 Producer 线程测试并行吞吐S3高并发多分区6 分区20 个 Producer 线程测试高并发S4弱副本集群ISR 仅 1 个副本模拟副本同步慢S5大消息压力消息 10KB测试带宽瓶颈S6持续写入稳定性连续写入 30 分钟观察延迟分布3.4 监控指标// 使用 Kafka Producer Metrics API 采集指标 MapMetricName, ? extends Metric metrics producer.metrics(); // 核心指标 double recordSendRate getMetric(metrics, record-send-rate); // 发送速率条/秒 double recordAckRate getMetric(metrics, record-ack-rate); // 确认速率条/秒 double requestLatencyAvg getMetric(metrics, request-latency-avg); // 平均请求延迟ms double requestLatencyMax getMetric(metrics, request-latency-max); // 最大请求延迟ms double recordQueueTimeAvg getMetric(metrics, record-queue-time-avg); // 队列等待时间 double ioWaitTimeNsAvg getMetric(metrics, io-wait-time-ns-avg); // IO 等待时间 double batchSizeAvg getMetric(metrics, batch-size-avg); // 平均 batch 大小 double compressionRate getMetric(metrics, compression-rate-avg); // 压缩率四、压测数据对比4.1 场景 S1单分区串行指标acks0acks1acksall吞吐量 (msg/s)85,20062,10038,400吞吐量 (MB/s)83.260.637.5平均延迟 (ms)1.22.87.6P99 延迟 (ms)3.16.518.2P99.9 延迟 (ms)5.812.335.6CPU 使用率 (%)455258分析单分区场景下acksall吞吐量仅为acks0的 45%。因为单分区只有一个 Leader 处理写入acksall需要额外等待 2 个 follower 同步串行等待效应明显。4.2 场景 S2多分区并行6 分区 × 6 线程指标acks0acks1acksall吞吐量 (msg/s)512,000438,000325,000吞吐量 (MB/s)500.0427.7317.4平均延迟 (ms)1.83.26.8P99 延迟 (ms)4.58.115.3P99.9 延迟 (ms)8.215.628.4CPU 使用率 (%)626872分析多分区后差距缩小。acksall吞吐量达到acks0的 63%。因为 6 个分区分布在 3 个 Broker 上follower 同步可以并行进行。4.3 场景 S3高并发多分区6 分区 × 20 线程指标acks0acks1acksall吞吐量 (msg/s)1,180,0001,020,000815,000吞吐量 (MB/s)1,152.3996.1795.9平均延迟 (ms)4.26.511.3P99 延迟 (ms)12.618.432.1P99.9 延迟 (ms)25.338.756.8CPU 使用率 (%)788285关键发现高并发下acksall突破百万级吞吐815K msg/s差距进一步缩小到 69%。原因是高并发下 batch 聚合更充分网络往返成本被均摊。4.4 场景 S4弱副本集群ISR 同步慢模拟 follower 落后场景将其中一个 Broker 的replica.fetch.max.bytes限制为 1KB/s。指标acks0acks1acksall吞吐量 (msg/s)510,000435,0008,200平均延迟 (ms)1.83.13,650P99 延迟 (ms)4.68.028,000超时率 (%)00.0142.5重要结论当副本同步异常时acksall性能断崖式下降。吞吐量从 325K 跌到 8K延迟从 7ms 飙到 3.6 秒42.5% 的请求超时。这就是为什么min.insync.replicas和 ISR 监控至关重要。4.5 场景 S5大消息10KB指标acks0acks1acksall吞吐量 (msg/s)85,00072,00051,000吞吐量 (MB/s)830.1703.1498.0平均延迟 (ms)5.27.814.5P99 延迟 (ms)15.322.138.6分析大消息场景下acksall的性能比例60%与小消息多分区场景63%接近。瓶颈转移到网络带宽上。4.6 场景 S630 分钟持续写入稳定性6 分区 × 10 线程持续 30 分钟每 1 分钟采样一次指标acks0acks1acksall平均吞吐量 (msg/s)980,000860,000690,000吞吐量标准差12,00018,00045,000最大延迟 (ms)3552180延迟波动范围 (ms)1-352-525-180关键发现acksall不仅平均延迟更高延迟波动也更大。标准差是acks0的 3.75 倍。这是因为 follower 的 Fetch 请求存在调度抖动偶发的慢拉取会拉长整个 batch 的确认时间。4.7 综合对比汇总五、延迟来源拆解acksall比acks0多出的延迟到底花在哪里5.1 延迟拆解模型延迟组件acks0acks1acksall说明队列等待1.2 ms1.2 ms1.2 msbatch 聚合时间网络发送0.5 ms0.5 ms0.5 msProducer → LeaderLeader 写入0.3 ms0.3 ms0.3 msLeader 日志追加副本同步——4.0 msFollower Fetch 写入网络响应0.2 ms0.5 ms0.5 msBroker → Producer总计2.2 ms2.5 ms6.5 ms副本同步是acksall延迟的主要来源占总延迟的 61.5%。5.2 副本同步为什么需要 4msFollower 同步 Leader 数据是通过Fetch 请求轮询的不是 Leader 主动推送// Follower 的 Fetch 线程ReplicaFetcherThread // 默认每 500ms 轮询一次replica.fetch.wait.max.ms while (true) { // 1. 向 Leader 发送 FetchRequest val fetchRequest buildFetchRequest(partitionMap) val response leaderBroker.fetch(fetchRequest) // 2. 将拉取到的数据写入本地日志 response.records.foreach { records localLog.append(records) } // 3. 更新本地 High Watermark updateHighWatermark() // 4. 等待下一轮默认 500ms 间隔 Thread.sleep(fetchWaitMaxMs) // replica.fetch.wait.max.ms }4ms 的分解Follower Fetch 请求到达 Leader 0.3 ms Leader 返回数据网络传输 0.5 ms Follower 写入本地日志 0.3 ms 等待下一轮 Fetch 轮询平均等待半个周期 2.5 ms ← 最大开销 Leader 检测 HW 更新并回调 0.4 ms关键优化减少 Fetch 轮询间隔可以显著降低延迟# Broker 端配置 replica.fetch.wait.max.ms100 # 从 500ms 降到 100ms默认 500 replica.fetch.min.bytes1 # 有数据立即拉取默认 1 byte replica.fetch.max.bytes1048576 # 单次拉取最大 1MB调整后重新压测 S3 场景指标默认配置优化配置改善平均延迟 (ms)11.37.2-36%P99 延迟 (ms)32.121.5-33%吞吐量 (msg/s)815,000845,0003.7%六、生产环境配置建议6.1 acks 选型决策表业务场景推荐 acks理由日志收集/监控指标0或1容忍少量丢数据追求吞吐用户行为埋点1偶尔丢几条不影响分析订单/支付流水all不能丢数据审计/合规日志all必须不丢实时推荐特征1容忍秒级数据丢失IoT 设备告警all告警不能丢CDN 访问日志0量大且非关键消息通知/IMall消息不能丢6.2 与 acksall 配套的必选配置# Producer 端 acksall retries2147483647 # 无限重试Integer.MAX_VALUE max.in.flight.requests.per.connection5 # 保障幂等性需 ≤5 enable.idempotencetrue # 开启幂等生产者 request.timeout.ms30000 # 30s 超时 delivery.timeout.ms120000 # 2min 投递超时 compression.typelz4 # 压缩减少网络开销 # Broker 端 min.insync.replicas2 # 至少 2 个副本同步成功 default.replication.factor3 # 3 副本 replica.fetch.wait.max.ms100 # 缩短 Fetch 间隔 replica.lag.time.max.ms10000 # 10s 未同步则移出 ISR unclean.leader.election.enablefalse # 禁止非 ISR 副本成为 Leader6.3min.insync.replicas的陷阱Topic: 3 副本, min.insync.replicas2 正常情况ISR[0,1,2], 3 个副本在线 → acksall 需要 3 个全部写入正常返回 一个副本宕机ISR[0,1], 2 个副本在线 → min.insync.replicas2满足条件acksall 仍可用 两个副本宕机ISR[0], 1 个副本在线 → min.insync.replicas2不满足条件 → acksall 直接返回 NotEnoughReplicasException → Topic 不可写入这就是可靠性 vs 可用性的 trade-off。min.insync.replicas2保证至少 2 个副本有数据但如果 2 个副本都宕机Topic 就不可写了。6.4 幂等生产者与 acksall// 幂等生产者Kafka 0.11 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等生产者自动设置 // acks all // retries Integer.MAX_VALUE // max.in.flight.requests.per.connection ≤ 5 // 幂等性通过 PID (Producer ID) Sequence Number 实现 // Producer 发送PID42, Seq0 → Broker 写入 [PID42, Seq0] // Producer 超时重试PID42, Seq0 → Broker 发现重复跳过写入 // Producer 发送下一条PID42, Seq1 → Broker 写入 [PID42, Seq1]开启幂等性后acksall的重试不会产生重复消息这是生产环境的标准配置。七、常见踩坑案例坑 1acksall 但 min.insync.replicas1# 危险配置 acksall min.insync.replicas1 # ← 问题在这里min.insync.replicas1意味着只要 Leader 自己写成功就算全部 ISR 同步完。如果 Leader 写完后立即宕机follower 还没拉取到数据消息就丢了。正确配置min.insync.replicas至少为 2。坑 2acksall 但 unclean.leader.electiontrueacksall unclean.leader.election.enabletrue # ← 问题在这里unclean.leader.election.enabletrue允许非 ISR 中的副本成为 Leader。这意味着一个落后很多的副本可能成为新 Leader导致已确认的消息丢失。正确配置unclean.leader.election.enablefalse。坑 3高并发下延迟飙升现象acksall在并发量上去后P99 延迟从 30ms 飙到 500ms。原因linger.ms0导致每个小消息都单独发送大量小请求挤满 Broker 的请求队列。修复linger.ms5 # 等待 5ms 凑 batch batch.size65536 # 64KB batch 大小 compression.typelz4 # 压缩减少请求体坑 4跨机房部署延迟放大两地三中心部署Leader 和 Follower 在不同机房网络 RTT 15ms。acksall延迟从 7ms 飙到 22ms增加了 15ms 跨机房 RTT。解决方案使用Conflent Multi-Region Clusters方案Leader 选举优先在本地机房或使用MirrorMaker2做跨机房异步同步本地集群用acks1八、总结维度acks0acks1acksall吞吐量百万级100%86%69%平均延迟最低50%170%P99 延迟最低46%154%延迟稳定性好中差数据可靠性最低中最高适用场景日志/指标埋点/特征订单/支付核心结论acksall 不是洪水猛兽在高并发 多分区场景下吞吐量只比 acks0 低 31%延迟增加约 7ms但前提是 ISR 健康一旦副本同步异常acksall 性能断崖式下降325K → 8K msg/s必须配套使用min.insync.replicas2unclean.leader.electionfalseenable.idempotencetrue延迟可优化缩短replica.fetch.wait.max.ms可将延迟降低 36%按业务选 acks不要全局一刀切按 Topic 可靠性需求分别配置下一篇预告本专栏下一篇文章将深入Spark 3.5 AQE自适应查询执行原理与 10 个生产调优案例——为什么同一份 Spark SQL开 AQE 后性能差 3 倍如果觉得有帮助点个赞和收藏关注专栏「AI大模型大数据硬件编程」不错过后续更新。
