使用observeOn时,为什么我的RxJava Flowable不支持背压？

我正在尝试创建一个Flowable,它发出关于背压的事件以避免内存问题,同时并行运行每个转换阶段以提高效率.我创建了一个简单的测试程序来推断我的程序的不同步骤的行为以及何时发出事件而不是在不同阶段等待.

我的计划如下：

public static void main(String[] args) throws ExecutionException, InterruptedException {
  Stream<Integer> ints = IntStream.range(0, 1000).boxed().collect(Collectors.toList())
      .stream().map(i -> {
        System.out.println("emitting:" + i);
        return i;
      });

  Flowable<Integer> flowable = Flowable.fromIterable(() -> ints.iterator());
  System.out.println(String.format("Buffer size: %d", flowable.bufferSize()));

  Long count = flowable.onBackpressureBuffer(10)
      .buffer(10)
      .flatMap(buf -> {
        System.out.println("Sleeping 500 for batch");
        Thread.sleep(500);
        System.out.println("Got batch of events");
        return Flowable.fromIterable(buf);
      }, 1)
      .map(x -> x + 1)
      .doOnNext(i -> {
        System.out.println(String.format("Sleeping : %d", i));
        Thread.sleep(100);
        System.out.println(i);
      })
      .count()
      .blockingGet();

  System.out.println("count: " + count);
}

当我运行它时,我得到的输出符合预期的背压,其中一批事件被激活到缓冲区中的大小,然后它们被平面映射,最后在逐个打印的地方采取一些操作：

Buffer size: 128
emitting:0
emitting:1
emitting:2
emitting:3
emitting:4
emitting:5
emitting:6
emitting:7
emitting:8
emitting:9
Sleeping 500 for batch
Got batch of events
Sleeping : 1
1
Sleeping : 2
2
Sleeping : 3
3
Sleeping : 4
4
Sleeping : 5
5
Sleeping : 6
6
Sleeping : 7
7
Sleeping : 8
8
Sleeping : 9
9
Sleeping : 10
10
emitting:10
emitting:11
emitting:12
emitting:13
emitting:14
emitting:15
emitting:16
emitting:17
emitting:18
emitting:19
Sleeping 500 for batch
Got batch of events
Sleeping : 11
11
Sleeping : 12
12
Sleeping : 13

但是,如果我尝试通过添加对.observeOn(Schedulers.computation())的一些调用来并行化这里的不同操作阶段,那么我的程序似乎不再尊重背压.我的代码现在看起来像：

public static void main(String[] args) throws ExecutionException, InterruptedException {
  Stream<Integer> ints = IntStream.range(0, 1000).boxed().collect(Collectors.toList())
      .stream().map(i -> {
        System.out.println("emitting:" + i);
        return i;
      });

  Flowable<Integer> flowable = Flowable.fromIterable(() -> ints.iterator());
  System.out.println(String.format("Buffer size: %d", flowable.bufferSize()));

  Long count = flowable.onBackpressureBuffer(10)
      .buffer(10)
      .observeOn(Schedulers.computation())
      .flatMap(buf -> {
        System.out.println("Sleeping 500 for batch");
        Thread.sleep(500);
        System.out.println("Got batch of events");
        return Flowable.fromIterable(buf);
      }, 1)
      .map(x -> x + 1)
      .observeOn(Schedulers.computation())
      .doOnNext(i -> {
        System.out.println(String.format("Sleeping : %d", i));
        Thread.sleep(100);
        System.out.println(i);
      })
      .observeOn(Schedulers.computation())
      .count()
      .blockingGet();

  System.out.println("count: " + count);
}

我的输出如下,其中我的所有事件都是预先发出的,而不是尊重各个执行阶段指定的背压和缓冲：

Buffer size: 128
emitting:0
emitting:1
emitting:2
emitting:3
emitting:4
emitting:5
emitting:6
emitting:7
emitting:8
emitting:9
emitting:10
Sleeping 500 for batch
emitting:11
emitting:12
... everything else is emitted here ...
emitting:998
emitting:999
Got batch of events
Sleeping 500 for batch
Sleeping : 1
1
Sleeping : 2
2
Sleeping : 3
3
Sleeping : 4
4
Sleeping : 5
Got batch of events
Sleeping 500 for batch
5
Sleeping : 6
6
Sleeping : 7
7
Sleeping : 8
8
Sleeping : 9
9
Sleeping : 10
Got batch of events
Sleeping 500 for batch
10
Sleeping : 11
11
Sleeping : 12
12
Sleeping : 13
13
Sleeping : 14
14
Sleeping : 15
Got batch of events
Sleeping 500 for batch
15
Sleeping : 16
16
Sleeping : 17
17
Sleeping : 18
18
Sleeping : 19
19
Sleeping : 20
Got batch of events
Sleeping 500 for batch
20
Sleeping : 21
21
Sleeping : 22
22
Sleeping : 23
23
Sleeping : 24
24
Sleeping : 25
Got batch of events
Sleeping 500 for batch
25

假装我的批处理阶段正在调用外部服务,但我希望它们因延迟而并行运行.我还希望在给定时间控制内存中的项目数,因为最初发出的项目数量可能是高度可变的,并且在批次上运行的阶段比初始事件发射要慢得多.

如何在调度程序中获得Flowable尊重背压？为什么我在调用observeOn时只会不尊重背压？

解决方法:

How can I have my Flowable respect backpressure across a Scheduler

实际上,应用onBackpressureBuffer会使其上方的源与下游应用的任何背压断开连接,因为它是一个无界限运算符.你不需要它,因为Flowable.fromIterable(顺便说一句,RxJava有一个范围运算符)支持并尊重背压.

Why does it seem to only disrespect backpressure when I sprinkle in calls to observeOn?

在第一个例子中,发生了一种称为call-stack blocking的自然背压.默认情况下,RxJava是同步的,并且大多数运算符不会引入异步,就像第一个示例中没有一样.

observeOn因此在理论上引入了异步边界,阶段可以彼此并行运行.它有一个默认的128元素预取缓冲区,可以通过其中一个重载进行调整.但是,在您的情况下,缓冲区(10)实际上会将预取量放大到1280,这仍然可能导致一次性完全消耗1000个元素的长源.

标签：java,multithreading,rx-java,reactivex
来源： https://codeday.me/bug/20190627/1306728.html