主线程完成后，Reactor Schedulers 继续运行很长时间？如何处理？

Question

我有一个关于如何在使用 Reactor 3 时清理调度程序工作线程的问题

Flux.range(1, 10000)
.publishOn(Schedulers.newElastic("Y"))
.doOnComplete(() -> { 
    // WHAT should one do to ensure the worker threads are cleaned up
    logger.info("Shut down all Scheduler worker threads");
})
.subscribe(x -> logger.debug(x+ "**"));

当我执行上面的代码时，我看到的是，一旦主线程完成运行，工作线程在一段时间内仍处于 WAITING 状态。

sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1081)
java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
java.lang.Thread.run(Thread.java:748)

有没有办法控制这个？也就是说，它们可以被处理掉onComplete()吗？我试过了Schedulers.shutdownNow()，但没有帮助。

另一方面，当我这样做时，我能够控制调度程序的处理。哪种是首选/提倡的方式？

reactor.core.scheduler.Scheduler s = Schedulers.newElastic("X");
        Flux.range(1, 10000)
        .concatWith(Flux.empty())
        .publishOn(s)
        .doOnComplete(() -> {           
            s.dispose();
            logger.info("Shut down all Scheduler worker threads");
        })
        .subscribe(x -> logger.debug(x+ "**"));

Answer 1

如果您使用Schedulers.new[Elastic|...]，那么您有责任跟踪结果Scheduler是否要关闭它。Schedulers.shutdownNow()只会在您不明确时关闭库使用的默认调度程序，例如Schedulers.elastic()（注意没有new前缀）。

在所有操作运行后进行清理的最佳方法是使用doFinally. 这将在 onError|之后异步执行清理回调。onComplete| cancel事件。更好地确保它是链中的最后一个操作符，尽管它试图在所有情况下真正最后执行。

唯一需要注意的是，它运行在同一个线程中以前的运营商，换句话说，非常线程你试图关闭......一个s.dispose()在doFinally回调将关机执行后，其任务的队列已经被处理，所以在这如果线程消失之前会有一点延迟。

这是一个转储线程信息的示例，切换到自定义弹性线程并将其关闭doFinally（添加过滤器并具体化以提供更短的日志，更好地了解事件如何进行）：

@Test
public void schedulerFinallyShutdown() throws InterruptedException {
    ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
    Logger logger = Loggers.getLogger("foo");
    CountDownLatch latch = new CountDownLatch(1);
    reactor.core.scheduler.Scheduler s = Schedulers.newElastic("X");
    Flux.range(1, 10000)
        .publishOn(s)
        .concatWith(Flux.<Integer>empty().doOnComplete(() -> {
            for (ThreadInfo ti : threadMxBean.dumpAllThreads(true, true)) {
                System.out.println("last element\t" + ti.getThreadName() + " " + ti.getThreadState());
            }
        }))
        .doFinally(sig -> {
            s.dispose();
            logger.info("Shut down all Scheduler worker threads");
            latch.countDown();
        })
        .filter(x -> x % 1000 == 0)
        .materialize()
        .subscribe(x -> logger.info(x+ "**"));

    latch.await();
    for (ThreadInfo ti : threadMxBean.dumpAllThreads(true, true)) {
        System.out.println("after cleanup\t" + ti.getThreadName() + " " + ti.getThreadState());
    }
} 

这打印出来：

11:24:36.608 [X-2] INFO  foo - onNext(1000)**
11:24:36.611 [X-2] INFO  foo - onNext(2000)**
11:24:36.611 [X-2] INFO  foo - onNext(3000)**
11:24:36.612 [X-2] INFO  foo - onNext(4000)**
11:24:36.612 [X-2] INFO  foo - onNext(5000)**
11:24:36.612 [X-2] INFO  foo - onNext(6000)**
11:24:36.612 [X-2] INFO  foo - onNext(7000)**
11:24:36.613 [X-2] INFO  foo - onNext(8000)**
11:24:36.613 [X-2] INFO  foo - onNext(9000)**
11:24:36.613 [X-2] INFO  foo - onNext(10000)**
last element    X-2 RUNNABLE
last element    elastic-evictor-1 TIMED_WAITING
last element    Monitor Ctrl-Break RUNNABLE
last element    Signal Dispatcher RUNNABLE
last element    Finalizer WAITING
last element    Reference Handler WAITING
last element    main WAITING
11:24:36.626 [X-2] INFO  foo - onComplete()**
11:24:36.627 [X-2] INFO  foo - Shut down all Scheduler worker threads
after cleanup   Monitor Ctrl-Break RUNNABLE
after cleanup   Signal Dispatcher RUNNABLE
after cleanup   Finalizer WAITING
after cleanup   Reference Handler WAITING
after cleanup   main RUNNABLE