books

《Java 消息中间件·可靠消息与消费治理》5.3 消费速率控制与背压

下游只有每秒 870 条的能力，消费者硬拉会怎样？实测不限速时 3.7 秒把 3 万条全拉进 JVM，队列峰值 29783，客户端内 p99 滞留 29.7 秒；用 pause/resume 把队列压到 1995，p99 降到 1.9 秒。附一份可运行的背压实现与 AIMD 限速器。

Leeting Yan 7 分钟阅读 3367 字

硬拉不会让下游变快

Kafka 消费者的默认行为是「能拉多少拉多少」：只要 broker 有数据、客户端有内存，poll() 就会把记录一批批搬进进程。下游慢的时候，这些记录不会消失，只会从 Kafka 的磁盘挪到 JVM 的堆里排队。积压没有减少，只是换了个地方，而且换到了一个更贵、更容易出事的地方。

本篇用一组对照实验说明这件事，然后给出一份可运行的 pause/resume 背压实现，以及一个按下游延迟自适应的限速器。

实验设置

环境：Temurin 25.0.5+8、kafka-clients 4.3.0、apache/kafka:4.3.0 单节点 KRaft，2026-10-29。单次运行，随调度波动，不作基准。

  • 主题 library.loan.returned.bp，1 分区，预灌 30000 条；
  • 下游用固定 1000 微秒/条模拟慢服务，理论能力约 1000 条/秒；
  • 两种模式的下游代码完全相同，只有「拉取是否受限速控制」不同；
  • 统计指标：拉取完成时间、队列峰值、记录从 poll() 返回到处理完成的滞留时间（p50/p99）。

这里的延迟只算客户端内部排队时间，不含记录在 Kafka 里等待的时间。Kafka 侧的等待就是 lag，属于另一个指标，5.1 已经拆过。

不限速：3.7 秒拉完，排队 30 秒

消费者用无界队列，poll() 拿到就 offer，下游线程按 1000 微秒/条消费。输出：

fetched=5000  processed=8   queue=4991  lag=25000
fetched=10451 processed=51  queue=10399 lag=19549
fetched=20423 processed=96  queue=20326 lag=9577
fetched=30000 processed=112 queue=29887 lag=0
result mode=unlimited fetched=30000 processed=30000 elapsedSec=34.0
       fetchAllSec=3.7 fetchPhaseRate=8109/s processRate=881/s
       maxQueue=29783 p50Ms=14987 p99Ms=29734

四个事实：

  1. 3.7 秒就把 30000 条全部拉进 JVM，拉取速率 8109 条/秒；
  2. 拉完时下游只处理了 112 条，队列里压着 29887 条；
  3. 总耗时 34 秒，和下游能力一致（30000 / 881 ≈ 34 秒），拉得快并没有让处理变快；
  4. 客户端内 p99 滞留 29.7 秒，最后一条记录在堆里等了将近 30 秒才被处理。

如果这时进程崩溃，29887 条未提交的记录会全部重新投递，这一轮拉取和排队白做；如果堆不够大，还会先 OOM。硬拉唯一「赢」的是 lag 数字：它在 3.7 秒内就归零了，看起来像问题解决了。

限速：拉取跟着下游走

同样 30000 条，改成有界队列加高低水位 pause/resume：

paused  queue=1995 processed=4     fetched=2000  lag=28000
resumed queue=362  processed=1637  fetched=2000  lag=28000
paused  queue=1856 processed=1643  fetched=3500  lag=26500
resumed queue=369  processed=3130  fetched=3500  lag=26500
...
fetched=30000 processed=28543 queue=1456 lag=0
result mode=bounded fetched=30000 processed=30000 elapsedSec=34.6
       fetchAllSec=32.4 fetchPhaseRate=925/s processRate=868/s
       maxQueue=1995 p50Ms=1114 p99Ms=1887

对照表：

指标不限速限速
拉完 30000 条耗时3.7 秒32.4 秒
拉取阶段速率8109 条/秒925 条/秒
下游处理速率881 条/秒868 条/秒
队列峰值297831995
客户端内 p50 滞留14987 ms1114 ms
客户端内 p99 滞留29734 ms1887 ms
总耗时34.0 秒34.6 秒

三个结论：

  • 吞吐不变。 总耗时由下游能力决定，两种模式都在 34 秒左右。背压不提升吞吐，它改变的是「排队排在哪」。
  • 延迟大幅下降。 p99 从 29.7 秒降到 1.9 秒，因为队列最多只存 2000 条，等 2000/870 ≈ 2.3 秒。
  • lag 归零更晚。 限速模式下 lag 在 32 秒才归零，比不限速晚得多。这是正确的代价：把积压留在 Kafka（可重放、有保留期、不占堆）比堆在 JVM 里安全。

限速模式日志里还能直接看到 pause 的效果：fetched=2000 在 paused 到 resumed 之间没有变化，lag 也停在 28000。pause 期间消费者不再拉新数据，但下游线程继续消化队列，队列从 1995 降到 362，然后触发 resume。

pause/resume 的语义

4.3.0 的 KafkaConsumer.pause 注释写得很清楚：

Suspend fetching from the requested partitions. Future calls to {@link #poll(Duration)} will not return
any records from these partitions until they have been resumed using {@link #resume(Collection)}.
Note that this method does not affect partition subscription. In particular, it does not cause a group
rebalance when automatic assignment is used.

Note: Rebalance will not preserve the pause/resume state.

四个要点：

  1. 它停的是 fetch，不是心跳。暂停期间仍然必须调用 poll()，否则消费者会被踢出组、或者超过 max.poll.interval.ms。
  2. 它不影响订阅关系，不会因为 pause 触发再平衡。
  3. 再平衡不保留 pause 状态。 分区被回收再分配后，暂停标记丢失，新分配的分区默认是运行状态。再平衡监听器里要重置本地状态。
  4. 底层的判断在 SubscriptionState 的 isFetchable()：
private boolean isFetchable() {
    return !paused && !pendingRevocation && !pendingOnAssignedCallback && hasValidPosition();
}

paused 的分区不会被加入下一次 fetch 请求，所以 position 也不会前进。这正是实验里 fetched 冻结的原因。

一份可运行的 pause/resume 实现

import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public final class BackpressurePoller<K, V> {

    private static final int HIGH_WATER = 1600;
    private static final int LOW_WATER = 400;

    private final KafkaConsumer<K, V> consumer;
    private final BlockingQueue<ConsumerRecord<K, V>> queue;
    private boolean paused;

    public BackpressurePoller(KafkaConsumer<K, V> consumer, int capacity) {
        this.consumer = consumer;
        this.queue = new LinkedBlockingQueue<>(capacity);
    }

    /** 返回本次拉到的记录；暂停期间返回空批次，但心跳照常。 */
    public ConsumerRecords<K, V> pollOnce(Duration timeout) throws InterruptedException {
        int depth = queue.size();
        if (!paused && depth >= HIGH_WATER) {
            consumer.pause(consumer.assignment());
            paused = true;
        } else if (paused && depth <= LOW_WATER) {
            consumer.resume(consumer.assignment());
            paused = false;
        }

        ConsumerRecords<K, V> records = consumer.poll(timeout);
        for (ConsumerRecord<K, V> record : records) {
            queue.put(record);
        }
        return records;
    }

    public void onRebalanceRevoked() {
        paused = false;
    }
}

四个设计点：

  • 队列容量要大于高水位加一批在途记录。 高水位 1600、max.poll.records=500、容量 3000，留了 1400 条余量。如果容量只比高水位大一点，pause 生效前已经发出的 fetch 响应回来，put 会阻塞在 poll 线程里，背压变成了卡死。
  • 高低水位是滞回区间。 如果只用单个阈值，队列在阈值附近抖动时会反复 pause/resume。1600 停、400 起，中间 1200 条的缓冲让状态切换有明确方向。
  • pause 之后继续 poll。 代码里 pollOnce 无论如何都会调 poll(timeout)，暂停时拿到空批次。这是心跳和 max.poll.interval.ms 的要求。
  • 再平衡要重置状态。 paused 是本地变量，分区被回收后它没有意义；在 ConsumerRebalanceListener.onPartitionsRevoked 里调 onRebalanceRevoked，避免「以为还暂停着，其实分区已经换人」。

记录只入队一次；返回值仅用于计数和监控，调用方不要再处理一遍。这段代码和实验里的差别只有一个：实验为了打印状态，把 pause/resume 写在了主循环里；生产版把它们收进 pollOnce，调用方只需要「poll、入队、处理」。

自适应用户限速

pause/resume 是二值控制，适合「下游能力大致固定」的场景。如果下游能力随时间变化（缓存命中率、数据库负载、第三方限流），更好的做法是按观测到的延迟动态调速。一个够用的控制器是 AIMD：加性增、乘性减。

public final class AimdRateLimiter {

    private double permitsPerSecond;
    private double tokens;
    private long lastRefillNanos = System.nanoTime();

    public AimdRateLimiter(double initialPermitsPerSecond) {
        this.permitsPerSecond = initialPermitsPerSecond;
        this.tokens = initialPermitsPerSecond;
    }

    public void acquire() throws InterruptedException {
        while (true) {
            long sleepMs;
            synchronized (this) {
                refill();
                if (tokens >= 1.0) {
                    tokens -= 1.0;
                    return;
                }
                sleepMs = Math.max(1L,
                        (long) Math.ceil((1.0 - tokens) / permitsPerSecond * 1000));
            }
            Thread.sleep(sleepMs);
        }
    }

    /** 每个统计窗口调用一次：p99 超目标就降 20%，否则每秒加 20。 */
    public synchronized void observeLatency(long p99Millis, long targetMillis) {
        if (p99Millis > targetMillis) {
            permitsPerSecond = Math.max(20.0, permitsPerSecond * 0.8);
        } else {
            permitsPerSecond = Math.min(50_000.0, permitsPerSecond + 20.0);
        }
    }

    private void refill() {
        long now = System.nanoTime();
        double elapsedSeconds = (now - lastRefillNanos) / 1e9;
        tokens = Math.min(permitsPerSecond, tokens + elapsedSeconds * permitsPerSecond);
        lastRefillNanos = now;
    }
}

用法是每条记录处理前 acquire()，每个统计窗口把下游的 p99 喂给 observeLatency。三个参数要按业务定：

  • 目标延迟：不是越低越好。目标定得比下游正常 p99 还低，限速器会一路降到下限。用「SLO 的一半」作为起点比较稳。
  • 降速比例 0.8：乘性减，一次降 20%，避免把下游打崩后才反应。
  • 增速 20/秒：加性增，慢慢试探。增速太快会在「刚好过载」的点反复震荡。

AIMD 的优点是实现简单、行为可解释；缺点是需要一个可靠的延迟观测。如果延迟统计本身被排队污染（比如把客户端等待也算进去），控制器会自己把自己压低。测量点必须放在下游调用返回之后、入队之前。

下游慢时不要硬拉

硬拉的代价，按严重程度排：

  1. 内存与 GC。 3 万条 100 字节的记录不大，但真实消息常是几 KB，加上反序列化后的对象，队列峰值可能吃掉几个 GB。GC 压力反过来又拖慢处理，形成正反馈。
  2. 崩溃后的重复。 拉进来但没提交的记录，崩溃后全部重投。拉得越多，重复窗口越大，2.2 的幂等压力越大。
  3. 处理超时被踢出组。 如果在 poll 线程里处理，一次 poll() 返回太多记录，处理时间超过 max.poll.interval.ms，消费者会被踢出组，触发再平衡，分区回到别人手里，重复更多。
  4. 假象。 客户端 lag 在几秒内归零，监控看起来「已恢复」，实际上问题只是从 Kafka 转移到了进程内。

判断自己有没有在硬拉，看三个量：队列深度是否持续接近 max.poll.records × 分区数；客户端 lag 是否远小于 committed 口径 lag；p99 滞留时间是否随积压线性增长。三个中两个成立，就该加背压了。

四种控制手段的边界

手段控制什么生效速度适用
pause/resume是否发起 fetch下一轮 poll下游能力固定、需要硬边界
降低 max.poll.records单次返回条数立即处理重、怕超 max.poll.interval.ms
降低 fetch.max.bytes单次 fetch 字节数立即大消息、内存敏感
令牌桶 / AIMD每秒处理条数秒级下游能力波动、需要平滑

它们可以叠加：max.poll.records 限制单批上限，pause 提供硬背压，AIMD 在两者之间做平滑。不要只调一个参数就指望解决所有问题。

回到 libraryd

libraryd 的归还事件要更新库存、计算逾期费、发通知。库存更新是核心，逾期费和通知可以慢。背压的实现可以是：

  • 队列按「核心操作」和「可延后操作」分开，核心队列满时 pause；可延后队列满时这次通知直接跳过并记入待补清单，消息位移照常提交，事后从事件回放补算；
  • AIMD 的目标延迟取库存更新 p99 的一半，避免把数据库打到极限；
  • 再平衡时重置 pause 状态，并从已提交位移重新开始，重复部分交给 2.2 的幂等消费者。

这样 lag 会在一段时间内保持高位，但它是可见的、可恢复的；而堆内的 3 万条记录既不可见，也不可恢复。

阅读导航：上一节 5.2 积压处置：扩容、跳过与降级 ｜ 下一节 6.1 死信队列的设计

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收