Java SubmissionPublisher 怎样处理订阅者过慢

Java SubmissionPublisher 遇到订阅者处理过慢时，先不要急着把缓冲区调得很大。它为每个订阅者维护独立缓冲，真正要先做的选择是：消息不能丢就用 submit 让生产者等待；消息允许降级就用 offer，根据返回值统计丢弃并决定是否重试。

官方地址：https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/SubmissionPublisher.html

处理慢订阅者的最小配方是：让订阅者通过 request 控制需求量，让发布端在 submit 的等待和 offer 的丢弃之间做业务选择；不要把负数返回值当成延迟，而要把它当成实际丢弃数量。

先确认慢的是哪个订阅者

SubmissionPublisher 的缓冲是按订阅者独立计算的。一个订阅者消费很慢时，首先被填满的是它自己的缓冲，不代表其他订阅者也已经落后。缓冲容量只是上限，不能替代消费速率；订阅者如果没有持续调用 request，发布端迟早会遇到资源不足。

默认构造器会使用公共 ForkJoinPool，并采用 Flow.defaultBufferSize() 作为最大缓冲容量的起点。需要控制慢订阅者影响面时，可以显式指定执行器和每个订阅者的最大缓冲容量。

消息不能丢时用 submit 让生产者背压等待

submit 会把消息异步交给当前订阅者；当任一订阅者的资源不可用时，调用线程会不可中断地等待。这个选择适合订单状态、账务事件或必须保序的内部消息，但要把发布线程被拖慢纳入容量设计。

import java.util.concurrent.Executors;
import java.util.concurrent.SubmissionPublisher;

public class SubmitBackpressure {
    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        var publisher = new SubmissionPublisher(executor, 8);

        publisher.consume(message -> {
            // 模拟订阅者处理较慢，消费速度低于发布速度。
            try {
                Thread.sleep(80);
            } catch (InterruptedException ex) {
                // 恢复中断标记，让上层能够感知线程需要停止。
                Thread.currentThread().interrupt();
            }
            System.out.println("consume=" + message);
        });

        for (int i = 0; i < 20; i++) {
            // submit 返回当前订阅者中的最大滞后估计值。
            int lag = publisher.submit("event-" + i);
            System.out.println("maxLag=" + lag);
        }

        // 关闭发布者后不再接受新消息，最后释放自建执行器。
        publisher.close();
        executor.shutdown();
    }
}

返回的整数是“最大滞后估计值”，不是精确队列长度，也不是毫秒延迟。它至少包含刚提交的这条消息（存在订阅者时），适合做监控线索；不要用它做并发同步条件。

消息可以降级时用 offer 观察丢弃

offer 不会像 submit 一样无限等待。无超时重载会尽力投递；如果某些订阅者的资源不足，返回负数，负数的绝对值就是本次投递失败的订阅者数量。回调返回 true 时只会再尝试一次，不能把它理解成无限重试器。

import java.util.concurrent.Executors;
import java.util.concurrent.SubmissionPublisher;

public class OfferWithDropHandler {
    public static void main(String[] args) {
        var executor = Executors.newFixedThreadPool(2);
        var publisher = new SubmissionPublisher(executor, 4);

        publisher.consume(message -> {
            // 低速消费者故意拉长处理时间，便于观察缓冲压力。
            try {
                Thread.sleep(120);
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
        });

        int dropped = 0;
        for (int i = 0; i < 30; i++) {
            int status = publisher.offer("event-" + i, (subscriber, item) -> {
                // 只记录一次丢弃；回调返回 false，表示不再次尝试。
                System.out.println("drop=" + item);
                return false;
            });
            if (status < 0) {
                // 负数的绝对值表示本次被多少个订阅者丢弃。
                dropped += -status;
            }
        }

        System.out.println("dropped=" + dropped);
        publisher.close();
        executor.shutdown();
    }
}

如果业务不能丢消息，看到负数就不能只打日志后继续；应该切回阻塞式 submit、把消息写入持久队列，或者向对应订阅者发出错误信号。丢弃策略必须和消息语义绑定。

用限时 offer 给等待设置预算

无超时 offer 适合“立即尝试，失败就降级”。如果希望给慢订阅者一个短暂追赶窗口，可以使用带超时的重载。它最多等待指定时间；超时或当前线程被中断后，才调用丢弃处理器。

import java.util.concurrent.TimeUnit;

int status = publisher.offer(
    "event-42",
    20,
    TimeUnit.MILLISECONDS,
    (subscriber, item) -> {
        // 超过 20 毫秒仍没有资源时，记录并放弃本次重试。
        System.out.println("degrade=" + item);
        return false;
    }
);

if (status < 0) {
    // 负数表示有订阅者没有收到该消息，必须进入可观测降级路径。
    System.out.println("droppedSubscribers=" + (-status));
} else {
    // 非负值仍然只是最大滞后估计，不代表所有消息都已消费。
    System.out.println("estimatedMaxLag=" + status);
}

超时值应来自业务预算，而不是随意填一个很大的数字。发布线程本身有截止时间时，限时 offer 能避免被一个慢订阅者无限拖住；但它也意味着调用方必须接受丢弃、转存或重试的后续动作。

把 request、异常和关闭动作接完整

慢订阅者通常不是只因为 onNext 里执行了耗时逻辑，也可能是它请求得太少。Flow.Subscriber 只有在 onSubscribe 中请求数据后才会接收消息；可以先请求一个小窗口，再在消费到一半时补充请求。

import java.util.concurrent.Flow;

final class WindowedSubscriber implements Flow.Subscriber {
    private final int window;
    private int remaining;
    private Flow.Subscription subscription;

    WindowedSubscriber(int window) {
        this.window = window;
        this.remaining = window;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        // 先申请一个有限窗口，避免一次性放开全部需求。
        this.subscription = subscription;
        subscription.request(window);
    }

    @Override
    public void onNext(String item) {
        // 先处理当前消息，再在窗口耗尽前补充需求量。
        System.out.println("consume=" + item);
        if (--remaining == 0) {
            remaining = window;
            subscription.request(window);
        }
    }

    @Override
    public void onError(Throwable error) {
        // 订阅者异常应进入统一监控或补偿路径。
        error.printStackTrace();
    }

    @Override
    public void onComplete() {
        // 正常关闭后不再请求新消息。
    }
}

如果执行器在异步通知时抛出 RejectedExecutionException，或者丢弃回调自身抛出异常，发布方法会把异常重新抛出；这时不一定所有订阅者都收到当前消息，通常应调用 closeExceptionally 让生命周期进入明确的错误状态。

close() 返回时也不保证所有订阅者已经完成回调。如果业务要等待消费结束，应另外维护完成信号；不要把 close() 的返回当成“所有消息已处理完毕”。

常见误区与速查表

可以记成一句话：不能丢就让 submit 背压等待；可以降级就用 offer 读取负数丢弃计数；需要时间边界就用限时 offer；无论选择哪种方式，都要让订阅者正确调用 request，并为异常和关闭留出明确路径。

相关问题

把最大缓冲容量调大就能解决慢订阅者吗？

只能延后缓冲耗尽，不能改变生产和消费的长期速率差。容量越大，积压消息占用的内存和延迟也可能越高。

offer 返回正数是不是代表消息已经处理完成？

不是。非负返回值是最大滞后估计值，不是消费确认；若需要业务确认，要在订阅者处理完成后单独设计确认或持久化机制。

一个订阅者很慢会不会让所有订阅者都停住？

每个订阅者有独立缓冲，但发布方法要同时面对当前订阅者集合。使用 submit 时，任一订阅者资源不足都可能让发布线程等待；使用 offer 时则可按返回状态降级。