订阅者如何通过反应性回拉压力控制发布者？

Question

我有一个发布者，它的发布速度可能比订阅者处理数据的速度更快。为了解决这个问题，我开始背压。因为我不想丢弃任何数据，所以我使用反应性回拉压力。我的理解是，订阅者能够告诉发布者何时发布更多数据，如本文及以下段落所述。

出版者是一个可流动的系统，它以同步的方式并行工作，然后合并成一个连续的可流动系统。数据应该被缓冲最多10个元素，当缓冲区满时，Flowable不应该发布更多数据，并等待下一个请求。

订阅者是一个一次性订阅者，在开始时请求10项。每个消耗的项目都需要一些计算，然后将请求一个新项目。

我的MWE看起来像这样：

List<Integer> src = new ArrayList<>();
for (int i = 0; i < 200; i++) {
    src.add(i);
}
Flowable.fromIterable(src)
        .parallel(10, 1)
        .runOn(Schedulers.from(Executors.newFixedThreadPool(10)))
        .flatMap(i -> Single.fromCallable(() -> {
            System.out.println("publisher: " + i);
            Thread.sleep(200);
            return i;
        }).toFlowable())
        .sequential(1)
        .onBackpressureBuffer(10)
        .observeOn(Schedulers.newThread())
        .subscribeOn(Schedulers.newThread())
        .doOnError(Throwable::printStackTrace)
        .subscribeWith(new DisposableSubscriber<Integer>() {
            @Override
            protected void onStart() {
                request(10);
            }
            @Override
            public void onNext(Integer integer) {
                System.out.println("subscriber: " + integer);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                request(1);
            }
            @Override
            public void onError(Throwable t) {
            }
            @Override
            public void onComplete() {
            }
        });
try {
    Thread.sleep(1000000);
} catch (InterruptedException e) {
    e.printStackTrace();
}

我希望这段代码能做到以下几点：订阅者请求前10项。出版商出版了前10个项目。然后订阅者在onNext中进行计算，并请求发布者发布的更多项目。

实际发生的情况：起初，发布者似乎无限制地发布项目。在某个时候，例如，在发布了14个项目后，订阅者处理了它的第一个项目。在这种情况下，发布者继续发布项目。在大约30个发布项目后，抛出一个io.reactivex.exceptions.MisSingBackpressureExctive： Buffer已满，流结束。

我的问题是：我做错了什么？我怎么能让订阅者控制发布者是否以及何时发布数据？显然，我做错了一些可怕的事情。否则，预期不会与现实有如此大的差异。

上述MWE的输出示例：

publisher: 5
publisher: 7
publisher: 8
publisher: 0
publisher: 2
publisher: 6
publisher: 9
publisher: 3
publisher: 4
publisher: 1
publisher: 18
publisher: 17
publisher: 15
subscriber: 0
publisher: 11
publisher: 10
publisher: 19
publisher: 13
publisher: 14
publisher: 12
publisher: 16
publisher: 27
publisher: 28
publisher: 23
publisher: 21
publisher: 29
publisher: 20
publisher: 25
publisher: 22
publisher: 26
io.reactivex.exceptions.MissingBackpressureException: Buffer is full

Answer 1

我不是Rx方面的专家，但让我试试看<代码>观测（…）有自己的默认缓冲区大小128。所以，从一开始，它就会从上游请求比缓冲区能容纳的更多的数据。

observeOn（…）接受可选的缓冲区大小覆盖，但即使您提供了它，ParallelFlowable也将调用您的flatMap（…） 方法的频率比您想要的要高。我不是100%确定为什么，也许它有自己的内部缓冲，它在将rails合并回sequential时执行。

我认为你可以通过使用flatMap（…）更接近你想要的行为 而不是parralel（…） ，提供maxConcurrency参数。

要记住的另一件事是，您不想调用subscribeOn（…） -这意味着影响整个上游的可流动性。所以如果你已经在调用parallel（…）。runOn（…） ，它没有效果，否则效果将是意外的。

有了以上的武装，我认为这会让你更接近你所寻找的：

    List<Integer> src = new ArrayList<>();
    for (int i = 0; i < 200; i++) {
        src.add(i);
    }
    Scheduler scheduler = Schedulers.from(Executors.newFixedThreadPool(10));
    Flowable.fromIterable(src)
            .flatMap(
                    i -> Flowable.just( i )
                            .subscribeOn(scheduler) // here subscribeOn(...) affects just this nested Flowable
                            .map( __ -> {
                                System.out.println("publisher: " + i);
                                Thread.sleep(200);
                                return i;
                            } ),
                    10) // max concurrency
            .observeOn(Schedulers.newThread(), false, 10) // override buffer size
            .doOnError(Throwable::printStackTrace)
            .subscribeWith(new DisposableSubscriber<Integer>() {
                @Override
                protected void onStart() {
                    request(10);
                }
                @Override
                public void onNext(Integer integer) {
                    System.out.println("subscriber: " + integer);
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    request(1);
                }
                @Override
                public void onError(Throwable t) {
                }
                @Override
                public void onComplete() {
                }
            });
    try {
        Thread.sleep(1000000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }