我正在尝试创建一个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
,然后对它们进行平面映射,最后采取一些操作将它们一一打印:
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
尊重跨a的背压Scheduler
?为什么当我偶尔拨打电话时,它似乎只是不尊重背压observeOn
?