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(() -> {
    Thread.sleep(1);
    return "1_" + counter1.incrementAndGet();
}).subscribeOn(Schedulers.fromExecutor(executor1)));

ParallelFlux<String> flux2 = Flux.merge(source).parallel(1024).concatMap(item -> Mono.fromCallable(() -> {
    Thread.sleep(100);
    return "2_" + counter2.incrementAndGet();
}).subscribeOn(Schedulers.fromExecutor(executor1)));

如果你对你的代码这样做，那么你会开始看到更接近你似乎期望的结果，2_ 航行过去 1_ 尽管它休眠了 100 次只要：

...
2_17075
2_17076
1_863
1_864
2_17077
1_865
2_17078
2_17079
...

但是，请注意：

I'd like to achieve a balance between publishers (relative to the thread-pool sizes and processing time)

你不能在这里选择数字来平衡输出，至少不可靠或以任何有意义的方式 - 线程调度将是完全任意的。如果你想这样做，那么你可以使用 this variant of the subscribe method，允许你在订阅消费者上显式调用 request()。然后，这允许您通过仅请求您准备处理的尽可能多的元素来提供背压。

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

Project Reactor: Handling fast and slow publishers

java

reactive-programming

project-reactor

reactive