如何恢复算子此前使用的runOn调度器并获取当前Scheduler？

问题核心 #

1. 能否在Reactor算子内部获取当前正在使用的Scheduler？
2. Mono.fromFuture()绑定到AWS CRT Http Client的原生线程执行，后续算子复用该线程导致Thread.currentThread().contextClassLoader为空，需要优雅解决类加载器问题，或确定原Mono物化时的调度器。

解决方案 #

1. 最直接：提前捕获类加载器（推荐） #

不需要依赖线程上下文或调度器切换，在myFunction被调用时（此时线程拥有正确的类加载器）直接保存引用，后续算子中直接使用：

fun myFunction(): Mono<String> {
    // 提前捕获当前线程的类加载器（此时是应用正常线程，上下文正确）
    val appClassLoader = Thread.currentThread().contextClassLoader
    return Mono.just("example")
        .flatMap { value ->
            Mono.fromFuture {
                // 第三方库调用，运行在原生线程
            }
        }
        .map {
            val resource = appClassLoader.getResources("META-INF/services/blah_blah")
            resource.asSequence().first().toString()
        }
}

这个方法完全规避了线程切换带来的上下文丢失问题，逻辑简单且可靠。

2. 切回应用默认调度器 #

如果必须依赖线程上下文的类加载器，可以在fromFuture()之后通过publishOn()切回应用常用的调度器，比如Schedulers.boundedElastic()（Spring WebFlux中处理阻塞操作的默认调度器，自带正确的类加载器上下文）：

fun myFunction(): Mono<String> {
    return Mono.just("example")
        .flatMap { value ->
            Mono.fromFuture {
                // 第三方库调用，运行在原生线程
            }
        }
        // 切换回带正确类加载器的调度器
        .publishOn(Schedulers.boundedElastic())
        .map {
            val resource = Thread.currentThread().contextClassLoader.getResources("META-INF/services/blah_blah")
            resource.asSequence().first().toString()
        }
}

如果是Netty环境，也可以使用Schedulers.fromExecutor(nettyEventLoopGroup)绑定到Netty的事件循环线程，这类线程通常也带有正确的上下文。

3. 自定义类加载器绑定调度器 #

如果需要长期复用带特定类加载器的线程，可以创建自定义调度器，确保所有线程都绑定应用类加载器：

// 全局自定义调度器，初始化时绑定应用类加载器
val classLoaderScheduler = Schedulers.newBoundedElastic(
    corePoolSize = 10,
    maxPoolSize = 100,
    threadNamePrefix = "class-loader-aware",
    threadFactory = { thread ->
        thread.apply {
            contextClassLoader = Thread.currentThread().contextClassLoader
        }
    }
)

fun myFunction(): Mono<String> {
    return Mono.just("example")
        .flatMap { value ->
            Mono.fromFuture {
                // 第三方库调用，运行在原生线程
            }
        }
        .publishOn(classLoaderScheduler)
        .map {
            val resource = Thread.currentThread().contextClassLoader.getResources("META-INF/services/blah_blah")
            resource.asSequence().first().toString()
        }
}

关于获取当前Scheduler的说明 #

Reactor没有提供直接获取当前执行线程所属Scheduler的API，原因是：

• Scheduler是线程池的抽象，单个线程可能被多个Scheduler复用（比如共享线程池场景）
• 线程本身不会携带所属Scheduler的元数据

如果需要跟踪调度器，通常的做法是：

1. 在Mono创建时显式指定调度器并保存引用，后续通过publishOn()切回
2. 通过Mono.subscriberContext()传递调度器引用（适合复杂上下文场景）

示例：显式指定并复用调度器

fun myFunction(): Mono<String> {
    val targetScheduler = Schedulers.boundedElastic()
    return Mono.just("example")
        .subscribeOn(targetScheduler)
        .flatMap { value ->
            Mono.fromFuture {
                // 第三方库调用
            }
        }
        .publishOn(targetScheduler)
        .map {
            // 后续操作
        }
}

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