Java Flow Publisher 怎样实现背压协议

Java 的 Flow.Publisher 不应该把数据无条件推给订阅者。背压协议的关键是：Flow.Subscriber 先通过 Flow.Subscription.request(n) 声明自己愿意接收多少项，发布者再在许可范围内调用 onNext(item)。官方 API 入口是 https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/Flow.html。

把 request 理解成“还可以发送多少项”的需求量，把 onNext 理解成“已经交付一项数据”。Publisher 只有在需求量大于零时才发送；取消、完成和错误都要让这条订阅关系进入终止状态。

下面用一个内存中的整数列表实现最小 Publisher。示例只讲清 request 和 onNext，不把 Reactor、RxJava 或线程调度细节混进协议本身。

Publisher、Subscriber 与 Subscription 分别负责什么

Publisher 持有数据来源并接受订阅；Subscriber 消费数据并决定何时继续要数据；Subscription 把两者连起来，同时提供 request(long) 和 cancel()。订阅成功后，Publisher 先调用一次 onSubscribe，Subscriber 如果不 request，就不会收到后续的 onNext。

这也解释了“发布者速度太快”为什么不能只靠队列解决：队列只能暂存数据，request 才是订阅者明确给出的需求边界。多个订阅者时，每条订阅关系都应有自己的需求量和取消状态。

先实现只服务一个订阅者的 Subscription

实现背压时可以先把问题收窄到单订阅。状态至少包括数据游标、尚未使用的需求量和是否终止。request(n) 收到非正数时不能继续发送；否则调用方无法区分“没有需求”和“需求已经耗尽”。

import java.util.List;
import java.util.concurrent.Flow;

// 一个只服务单个订阅者的内存发布者，专门演示 request 与 onNext 的契约。
final class ListPublisher implements Flow.Publisher {
    private final List items;

    ListPublisher(List items) {
        // 复制列表，避免外部在订阅期间修改数据源。
        this.items = List.copyOf(items);
    }

    @Override
    public void subscribe(Flow.Subscriber super Integer> subscriber) {
        // 先建立订阅关系，再由订阅者在 onSubscribe 中发出需求量。
        subscriber.onSubscribe(new ListSubscription(subscriber));
    }

    private final class ListSubscription implements Flow.Subscription {
        private final Flow.Subscriber super Integer> subscriber;
        private int index;
        private long demand;
        private boolean terminated;

        ListSubscription(Flow.Subscriber super Integer> subscriber) {
            this.subscriber = subscriber;
        }

        @Override
        public synchronized void request(long n) {
            // Reactive Streams 约定：非正请求进入错误终止路径。
            if (n  Long.MAX_VALUE - n ? Long.MAX_VALUE : demand + n;
            emit();
        }

        private void emit() {
            // 每交付一项就消耗一份需求量，需求量归零立即停止发送。
            while (!terminated && demand > 0 && index 

这里的 emit() 是背压的核心：循环条件同时检查 demand > 0 和数据是否还有剩余。Long.MAX_VALUE 被当作“有效的近似无限需求”处理，但仍不能让普通业务默认使用它。

让 Subscriber 先申请一小批数据

订阅者不需要一次性申请全部数据。可以先申请 2 项，处理完后再申请下一批。这样，Publisher 每次调用 onNext 都对应一份已获得的需求量。

import java.util.concurrent.Flow;

// 按小批量消费数据，处理完当前批次后再补充需求量。
final class BatchSubscriber implements Flow.Subscriber {
    private final int batchSize;
    private Flow.Subscription subscription;
    private int remaining;

    BatchSubscriber(int batchSize) {
        this.batchSize = batchSize;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        // 保存订阅关系，并明确声明第一批可接收的数量。
        this.subscription = subscription;
        this.remaining = batchSize;
        subscription.request(batchSize);
    }

    @Override
    public void onNext(Integer item) {
        // 业务处理应在这里完成，不能把尚未处理的数据无限堆入本地队列。
        System.out.println("consume: " + item);
        if (--remaining == 0) {
            // 当前批次消费完，再向发布者申请下一批。
            remaining = batchSize;
            subscription.request(batchSize);
        }
    }

    @Override
    public void onError(Throwable throwable) {
        // 错误后不再 request，交给上层记录或触发恢复。
        System.err.println("flow failed: " + throwable.getMessage());
    }

    @Override
    public void onComplete() {
        // 数据耗尽后的正常终止点。
        System.out.println("flow complete");
    }
}

实际项目中，onNext 的处理如果是异步的，就不要在尚未完成处理时提前补充 request；否则本地缓存仍可能超过消费者能力。批大小是吞吐量和内存占用之间的工程取舍。

取消、完成和错误要互相排斥

一条订阅关系进入终止状态后，Publisher 不应再发送 onNext。正常耗尽调用 onComplete()；无效 request 或数据源异常调用 onError()；消费者主动停止则调用 cancel()。这三个出口都应设置终止标记，避免并发或重入导致重复通知。

示例的 request 使用了同步锁，是为了让需求量、游标和终止状态保持一致。更复杂的异步 Publisher 还要把串行通知、线程切换和异常传播一起设计，不能只把 emit() 放进线程池就认为已经完成背压。

批量 request 该怎么选

选择批量时先看单项处理耗时、可接受的内存和数据源是否可暂停。Flow 的默认缓冲建议可以作为起点，但不能替代业务测量；如果消费者处理速度经常低于生产速度，就应缩小 request 或增加真正受控的异步边界。

常见问题

Publisher 能在没有 request 时先发一项吗？

不应该。背压契约就是由 Subscriber 的需求量控制后续 onNext；没有需求量时继续发送会破坏订阅者的容量边界。

request(0) 是暂停的意思吗？

不是。非正数请求应按非法请求处理；要暂停发送，保留当前订阅关系并暂时不再 request，恢复时再申请正数。

request(Long.MAX_VALUE) 还算背压吗？

它表示近似无限需求，协议仍然有效，但实际上放弃了细粒度的消费控制。只有在资源可控、数据量明确或下游本身能承受时才适合使用。

为什么 onNext 后还要检查 cancel？

因为消费者可能在处理回调期间取消订阅。发送循环的下一轮必须重新检查终止状态，避免取消后继续交付后续数据。