问题描述
我正在尝试实现每个组的并行性,其中分组元素并行运行,并且在组内每个元素按顺序工作。但是对于下面的代码,第一个发射使用并行线程,但对于后续发射它使用一些不同的线程池。如何实现组内元素的组并行和顺序执行。
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,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
解决方法
暂无找到可以解决该程序问题的有效方法,小编努力寻找整理中!
如果你已经找到好的解决方法,欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@)