订阅者如何来控制发布者？
我有一个发布者，其发布速度可能比订阅者处理数据的速度快。为了解决这个问题，我开始使用背压。因为我不想丢弃任何数据，所以我使用了反应性拉背压。我理解这是因为订阅服务器可以告诉发布服务器何时发布更多数据，如本节及以下各段所述。
发布者是一个Flowable，它并行地异步执行其工作，然后合并到一个顺序的Flowable中。数据最多应缓冲10个元素，并且当此缓冲区已满时，Flowable不应再发布任何数据并等待下一个请求。
订户是一个DisposableSubscriber，它在开始时请求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.MissingBackpressureException: Buffer is full引发a，并且流结束。
我的问题：我在做什么错？如何让订阅者控制发布者是否以及何时发布数据？显然，我做错了什么。否则，期望与现实不会有太大不同。
上述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
问题来源：Stack Overflow
写回答
-
不是Rx方面的专家，但是让我at一息.. observeOn(...)它有自己的默认缓冲区大小128。因此，从一开始，它就会向上游请求更多的缓冲区。
observeOn(...)接受可选的缓冲区大小覆盖，但是即使您提供了它，ParallelFlowable也会flatMap(...)比您想要的更频繁地调用您的方法。我不是100％知道为什么，也许它有自己的内部缓冲，在将滑轨合并回顺序时会执行。
我认为您可以通过使用flatMap(...)而不是parralel(...)提供maxConcurrency参数来更接近所需的行为。
要记住的另一件事是您不想调用subscribeOn(...)-这是要完全影响上游Flowable。因此，如果您已经在调用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(); }回答来源：Stack Overflow
2020-03-22 15:17:19赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
