流量平面图背压

Question

如何应用背压来限制发行商生产比并行运行的flatMap更多的项目？

为了说明起见，这里有一个快速twitter用户名生成器、一个慢速twitter查找调用、一个慢速twitter文件编写器和一个打印方法。

private Consumer<FluxSink<String>> twitterUsernameGenerator() {
    return (sink) -> Stream.of("a", "b", "c", "d").forEach(sink::next);
}

private Flux<TwitterMessage>findTwitterMessagesByUsername() {
    return Flux.create(sink -> {
            sink.next(new TwitterMessage(...));
            sleep(2000);
            sink.next(new TwitterMessage(...));
        }
    });
}

private void print(Object o) {
    System.out.println("[" + Thread.currentThread().getName() + "] " + o);
}

最终目标是并行运行twitter查找，同时对生成器施加反压力，使其不会发出超出可处理范围的用户名（预计会有一些预取）。

Flux.create(twitterUsernameGenerator())
    .publishOn(Schedulers.single())
    .doOnNext(this::print)
    .subscribe();

这很好地在一个单独的线程上生成了5个twitter用户名

[single-1] a
[single-1] b
[single-1] c
[single-1] d

不确定它是正确的，但我的理由是，从一个用户名产生许多twitter消息和并行在两个线程上执行这种I/O密集型操作。

Flux.create(twitterUsernameGenerator())
    .publishOn(Schedulers.single())
    .doOnNext(this::print)
    .parallel(2)
    .runOn(Schedulers.newParallel("p", 2))
    .flatMap(username -> findTwitterMessagesByUsername(username))
    .doOnNext(this::print)
    .subscribe();

哇！生成器生成用户名的速度快于我们的处理速度。

[single-1] a
[single-1] b
[single-1] c
[single-1] d
[p-1] TwitterMessage{...}
[p-2] TwitterMessage{...}
...

我如何给发电机函数施加背压，这样结果就更接近这个了

[single-1] a
[single-1] b
[p-1] TwitterMessage{...}
[single-1] c
[p-2] TwitterMessage{...}
[single-1] d
...

Answer 1

背压发生在大于4个元素的“批次”中。如果修改生成器以生成更多用户名，例如。

  private Consumer<FluxSink<String>> twitterUsernameGenerator() {
        return (sink) -> IntStream.rangeClosed(0, 100000)
                .boxed().map(String::valueOf)
                .collect(Collectors.toList())
                .forEach(sink::next);
    }

由此产生的流量将表现出与预期流量类似的行为。

您可以在原始通量中添加onBackpressureError（）：

Flux.create(twitterUsernameGenerator())
        .onBackpressureError()
        .publishOn(Schedulers.single())
        .....

当出现反压力时，它将通过抛出异常清楚地发出信号。