具有并行性的 Reactor GroupBy 在同一线程上运行答案

【问题标题】：Reactor GroupBy with Parallelism Runs on same thread具有并行性的 Reactor GroupBy 在同一线程上运行
【发布时间】：2021-07-27 09:10:47
【问题描述】：

我正在尝试实现每个组的并行性，其中分组元素并行运行，并且在组内每个元素按顺序工作。然而对于下面的代码，第一个发射使用并行线程，但对于后续发射它使用一些不同的线程池。如何实现组内元素的组并行和顺序执行。

public class ReactorTest implements SmartLifecycle, ApplicationListener<ApplicationReadyEvent> {

    private AtomicInteger counter = new AtomicInteger(1);
    private Many<Integer> healthSink;
    private Disposable dispose;

    private ScheduledExecutorService executor;

    @Override
    public void start() {
        executor = Executors.newSingleThreadScheduledExecutor();
        healthSink = Sinks.many().unicast().onBackpressureBuffer();
        dispose = healthSink.asFlux().groupBy(v -> v % 3 == 0).parallel(10)
                .runOn(Schedulers.newBoundedElastic(10, 100, "k-task")).log().flatMap(v -> v)
                .subscribe(v -> log.info("Data {}", v));
    }

    @Override
    public void stop() {
        executor.shutdownNow();
        if (dispose != null) {
            dispose.dispose();
        }
    }

    @Override
    public boolean isRunning() {
        return executor == null ? false : !executor.isShutdown();
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {

        executor.scheduleAtFixedRate(() -> {
            healthSink.tryEmitNext(counter.incrementAndGet());
            healthSink.tryEmitNext(counter.incrementAndGet());
            healthSink.tryEmitNext(counter.incrementAndGet());
        }, 10, 10, TimeUnit.SECONDS);
    }
}

日志

2021-07-27 14:15:34.189  INFO 22212 --- [  restartedMain] i.g.kprasad99.reactor.DemoApplication    : Started DemoApplication in 1.464 seconds (JVM running for 1.795)
2021-07-27 14:15:44.206  INFO 22212 --- [       k-task-1] reactor.Parallel.RunOn.1                 : onNext(UnicastGroupedFlux)
2021-07-27 14:15:44.207  INFO 22212 --- [       k-task-2] reactor.Parallel.RunOn.1                 : onNext(UnicastGroupedFlux)
2021-07-27 14:15:44.207  INFO 22212 --- [       k-task-1] io.github.kprasad99.reactor.ReactorTest  : Data 2
2021-07-27 14:15:44.207  INFO 22212 --- [       k-task-2] io.github.kprasad99.reactor.ReactorTest  : Data 3
2021-07-27 14:15:44.207  INFO 22212 --- [       k-task-1] io.github.kprasad99.reactor.ReactorTest  : Data 4
2021-07-27 14:15:54.200  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 5
2021-07-27 14:15:54.200  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 6
2021-07-27 14:15:54.200  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 7
2021-07-27 14:16:04.195  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 8
2021-07-27 14:16:04.195  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 9
2021-07-27 14:16:04.195  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 10
2021-07-27 14:16:14.206  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 11
2021-07-27 14:16:14.206  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 12
2021-07-27 14:16:14.206  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 13
2021-07-27 14:16:24.197  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 14
2021-07-27 14:16:24.197  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 15
2021-07-27 14:16:24.197  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 16
2021-07-27 14:16:34.196  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 17
2021-07-27 14:16:34.196  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 18
2021-07-27 14:16:34.196  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 19
2021-07-27 14:16:44.201  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 20
2021-07-27 14:16:44.201  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 21
2021-07-27 14:16:44.201  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 22
2021-07-27 14:16:54.201  INFO 22212 --- [pool-3-thread-1] io.github.kprasad99.reactor.ReactorTest  : Data 23

【问题讨论】：

我认为原因之一是您使用的是Sinks.many().unicast()，而文档说它是Help building Sinks.Many that will broadcast signals to a single Subscriber。我尝试使用multicast()，但我不得不大量重构您的代码，因为它不能开箱即用。但是，一般来说，如果我使用简单的Flux（不是Disposable），即使没有.parallel(10)，代码也会在不同的线程上运行。
@Felipe 但我只有一个订阅者。此外，这只是一个示例，其中我在同一类中发布和订阅，但意图是从不同的地方发布。因为我认为它的单一订阅者我使用单播。

标签： spring spring-webflux project-reactor reactor-netty

【解决方案1】：

您需要将.parallel(..) 放在 .flatMap(..) 运算符之后：

Flux.interval(Duration.ofMillis(100))
  .take(10)
  .groupBy(v -> v % 2 == 0)
  .flatMap(f -> f)
  .parallel(2)
  .runOn(Schedulers.newBoundedElastic(2, 10, "k-task"))
  .subscribe(i -> log.info("Data {}", i));

结果：

10:32:33.377 [k-task-1] INFO  Data 0
10:32:33.466 [k-task-2] INFO  Data 1
10:32:33.562 [k-task-1] INFO  Data 2
10:32:33.673 [k-task-2] INFO  Data 3
10:32:33.766 [k-task-1] INFO  Data 4
10:32:33.860 [k-task-2] INFO  Data 5
10:32:33.971 [k-task-1] INFO  Data 6
10:32:34.065 [k-task-2] INFO  Data 7
10:32:34.163 [k-task-1] INFO  Data 8
10:32:34.268 [k-task-2] INFO  Data 9

【讨论】：