Project Reactor：处理快速和缓慢的发布者

Question

考虑以下代码：

AtomicInteger counter1 = new AtomicInteger();

AtomicInteger counter2 = new AtomicInteger();

Flux<Object> source = Flux.generate(emitter -> {

    emitter.next("item");

});

Executor executor1 = Executors.newFixedThreadPool(32);

Executor executor2 = Executors.newFixedThreadPool(32);

Flux<String> flux1 = Flux.merge(source).concatMap(item -> Mono.fromCallable(() -> {

        Thread.sleep(1);

        return "1_" + counter1.incrementAndGet();

}).subscribeOn(Schedulers.fromExecutor(executor1)));

Flux<String> flux2 = Flux.merge(source).concatMap(item -> Mono.fromCallable(() -> {

    Thread.sleep(100);

    return "2_" + counter2.incrementAndGet();

}).subscribeOn(Schedulers.fromExecutor(executor2)));

Flux.merge(flux1, flux2).subscribe(System.out::println);

您可以看到一个发布者比另一个发布者快 100 倍。不过，在运行代码时，似乎所有数据都已打印，但两个发布者之间存在巨大差距，这会增加加班时间。

有趣的是，当更改数字时executer2会有1024线程，并且executer1只有1线程，然后我们仍然会看到随着时间的推移越来越大的差距。

我的期望是，在相应地调整线程池之后，发布者将得到平衡。

我想在发布者之间取得平衡（相对于线程池大小和处理时间）

如果我等的时间足够长会发生什么？换句话说，是否会发生背压？（默认情况下，我猜这是一个运行时异常，对吧？）

我不想丢弃物品，也不想出现运行时异常。相反，正如我提到的，我希望系统在其拥有的资源和处理时间方面取得平衡——上面的代码是否承诺了这一点？

Answer 1

本例中的对象Flux不是ParallelFlux对象，因此它们只会使用一个线程。如果您创建一个能够处理数千个线程的调度程序，并将其传递给其中一个对象，这并不重要Flux——它们只会坐在那里不被使用，这正是本示例中发生的情况。没有背压，也不会导致异常——它会尽可能快地使用一个线程。如果要确保Flux充分利用可用的 1024 个线程，则需要调用.parallel(1024)：ParallelFlux<String> flux1 = Flux.merge(source).parallel(1).concatMap(item -> Mono.fromCallable(() -> {&nbsp; &nbsp; Thread.sleep(1);&nbsp; &nbsp; return "1_" + counter1.incrementAndGet();}).subscribeOn(Schedulers.fromExecutor(executor1)));ParallelFlux<String> flux2 = Flux.merge(source).parallel(1024).concatMap(item -> Mono.fromCallable(() -> {&nbsp; &nbsp; Thread.sleep(100);&nbsp; &nbsp; return "2_" + counter2.incrementAndGet();}).subscribeOn(Schedulers.fromExecutor(executor1)));如果你对你的代码这样做，那么你会开始看到更接近你似乎期望的结果，尽管它的休眠时间是原来的 100 倍，但还是会过去2_：1_...2_170752_170761_8631_8642_170771_8652_170782_17079...但是，请注意：我想在发布者之间取得平衡（相对于线程池大小和处理时间）你不能在这里选择数字来平衡输出，至少不能可靠地或以任何有意义的方式——线程调度将是完全任意的。如果你想这样做，那么你可以使用subscribe 方法的这个变体，允许你显式地调用request()订阅消费者。这样一来，您就可以通过仅请求您准备处理的尽可能多的元素来提供背压。