WebFlux中使用Schedulers.boundedElastic()时如何正确打印日志？

• Schedulers.boundedElastic()创建的线程均为守护线程，JVM进程退出时不会等待守护线程执行完成
• Junit测试主线程执行完subscribe()方法后会直接结束，此时异步任务还未执行到日志打印逻辑就被强制终止，因此没有日志输出
• 无subscribeOn时整条链路逻辑运行在Junit主线程，所有逻辑执行完测试方法才会退出，因此日志正常；添加Thread.sleep是主动让主线程等待异步线程执行，因此也能正常打印日志

测试场景（推荐使用StepVerifier） #

Reactor官方提供的测试工具，既保留你定义的异步调度逻辑，又能自动等待响应式流执行完成，无需手动加sleep：

import reactor.test.StepVerifier;

class Foo1 {
    private static final org.slf4j.Logger logger = LoggerFactory.getLogger(Foo1.class);

    @Test
    public void test1() {
        Mono<String> businessMono = Mono.just("Input Data").flatMap(msg -> {
            try {
                logger.info("Message input in logger");
                // 业务逻辑
                return Mono.just("Message processed");
            } catch (Exception e) {
                logger.error(e.toString());
                return Mono.just("Message not processed");
            }
        }).doOnNext(res -> logger.info(res))
          .subscribeOn(Schedulers.boundedElastic())
          .doOnSuccess(c -> logger.info("subscribe completed"));
        
        // 等待流执行完成，可同时验证执行结果
        StepVerifier.create(businessMono)
                .expectNext("Message processed")
                .verifyComplete();
    }
}

如果不需要验证执行结果，也可以直接调用block()方法阻塞等待执行完成，异步调度逻辑依然生效：

@Test
public void test1() {
    Mono.just("Input Data").flatMap(msg -> {
        try {
            logger.info("Message input in logger");
            // 业务逻辑
            return Mono.just("Message processed");
        } catch (Exception e) {
            logger.error(e.toString());
            return Mono.just("Message not processed");
        }
    }).doOnNext(res -> logger.info(res))
      .subscribeOn(Schedulers.boundedElastic())
      .subscribe(c -> logger.info("subscribe completed"))
      // 阻塞等待异步任务执行完成，仅测试场景使用
      .block();
}

生产常驻进程场景 #

不需要任何额外处理，只要服务进程不主动退出，守护线程的异步任务和日志都会正常执行打印，无需加sleep或者阻塞逻辑。

内容的提问来源于stack exchange，提问作者pri