PublishProcessor是如何实现的。offer（）知道下游没有消耗之前的排放量吗？

Question

我使用PublishProcessor。提供（）从上游排放。我读到（*），如果订阅者没有准备好接收下一个事件，它可能会返回false，因为PublishProcessor不协调背压。我想知道它是如何工作的。PublishProcessor如何知道订阅者还没有准备好？

我了解到上游可能知道下游通过反应性拉动处理排放的潜力：

someObservable.subscribe(new Subscriber<t>() {
    @Override
    public void onStart() {
      request(1);
    }

    @Override
    public void onCompleted() {
      // gracefully handle sequence-complete
    }

    @Override
    public void onError(Throwable e) {
      // gracefully handle error
    }

    @Override
    public void onNext(t n) {
      // do something with the emitted item "n"
      // request another item:
      request(1);
    }
});

https://github.com/ReactiveX/RxJava/wiki/Backpressure

但我的订阅看起来是这样的：

publishProcessor.subscribe(new Consumer<T>() {
                @Override
                public void accept(T t) throws Exception {
                    // do IO
                }
            }, Log::submitCrash);

是否在内部调用请求（1）；accept（）完成后，还是不执行反应性拉取？我试着读代码，但似乎不是这样。该消费者被传递给LambdaSubscriber。还有兰博达。onNext（）不调用请求（n）。

此订阅html" target="_blank">方法的javadoc：

 public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) 

说明：操作符以无限制的方式使用源{@code Publisher}（即，没有对其施加反压力）。

所以它没有提到反应性拉力。以无限制的方式使用源是否意味着对该订阅应用了无限制的背压缓冲区，或者根本没有背压缓冲区？

PublishProcessor是否还有其他机制可以知道其订阅者是否已经完成了对已发出值的消耗？

（*）

PublishProcessor既是一个可流动的处理器，也是一个可流动的处理器，但它不协调不同订户之间以及上游源和订户之间的背压。如果通过onNext（对象）接收上游项目，如果订阅者尚未准备好接收项目，则该订阅者将通过MissingBackpressureException终止。为了避免这种情况，请使用offer（Object），如果返回false，请稍后重试。

http://reactivex.io/RxJava/javadoc/io/reactivex/processors/PublishProcessor.html

Answer 1

每个订户都会收到一个特殊的订阅，该订阅跟踪消费者发出的请求金额，如果PublishProcessor可以在订户上调用onNext，则会减少该金额。

如果跟踪的金额为零，并在报价中检查，则认为消费者尚未准备好。

如果您同步使用PublishProcessor，并从onNext中发出请求，跟踪金额将始终大于零，因此报价可以通过。

但是，如果您使用异步，例如，通过应用观察，现在处理器和用户之间有一个有界缓冲区，该缓冲区可能会填满，跟踪的请求量可能最终为零，从而阻止offer发出更多信号。

无限消费意味着消费者发出请求（Long.MAX_VALUE），这被解释为消费者准备好接收任意数量的商品PublishProcessor本身会这样做，以防您将其订阅给另一家Publisher。

一般来说，unbounded并不意味着无限缓冲，因为特定消费者可能会在onNext的调用线程上同步丢弃、批处理、采样或处理项目，因此不会发生溢出。