登录
推荐 文章 Go 技术 课程 下载 专题 AI
首页 >  文章 >  java教程

Java SubmissionPublisher 怎样处理订阅者过慢

来源：17golang原创

时间：2026-11-06 20:45:58 205浏览 收藏

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() 作为最大缓冲容量的起点。需要控制慢订阅者影响面时，可以显式指定执行器和每个订阅者的最大缓冲容量。

SubmissionPublisher 为每个订阅者维护独立缓冲的结构示意图
图1：发布者、独立订阅者缓冲与消费请求之间的静态结构说明图，不是运行截图或运行证据。

消息不能丢时用 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、把消息写入持久队列，或者向对应订阅者发出错误信号。丢弃策略必须和消息语义绑定。

SubmissionPublisher 的 submit 与 offer 返回状态和慢订阅者处理策略对比图
图2：submit 等待、offer 丢弃与负数状态之间的关系结构图，不是软件截图。

用限时 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 且某个订阅者缓冲已满检查 request 速率、执行器容量和发布线程等待预算
offer 返回负数至少一个订阅者没有收到当前消息记录丢弃数量，按消息重要性转存、重试或降级
一直收不到 onNext订阅者没有 request，或已取消检查 onSubscribe、cancel 和订阅生命周期
close 后仍想发布发布者已停止接受提交在关闭前完成生产协调，关闭后不要复用实例

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

相关问题

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

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

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

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

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

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

声明：本文转载于：17golang原创 如有侵犯，请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>