PAYATHON 2026

如何按条件调用 .publishOn(boundedElastic())？

技术老齐

第二种设计在示例中是合理的：flatMap 收到上游值后，会在处理该 onNext 信号的线程上执行映射函数，再订阅内部的 Mono。把线程判断放在这里，就能根据实际收到信号时的线程决定是否调度。

要严格满足“必须在 boundedElastic 上执行”的要求，还得可靠地识别线程。getName().startsWith("boundedElastic-") 只能作为受控环境下的命名约定，不能证明线程属于某个 Scheduler。

为什么第一种方案判断错了线程

transformDeferred 的转换函数在每次订阅时执行。此时，上游可能还没产生信号，也还没经过后续的异步调度。

第一个示例的执行顺序是：

  1. main 调用 subscribe()。
  2. transformDeferred 在 main 上检查线程，决定添加 publishOn。
  3. 上游的 process 经由 subscribeOn 在 boundedElastic 上执行。
  4. 信号到达已经添加的 publishOn，再次调度后执行 send。

外层 transformDeferred 检查的是订阅时的线程，无法据此判断上游最终会在哪个线程发出值。

publishOn(Schedulers.boundedElastic()) 会建立调度边界。即使当前线程来自同一个 Scheduler，也不能假定它会直接在当前线程执行，具体使用哪个编号的线程也没有保证。

在 postProcess 中按信号判断

第二种方案中的 Mono.fromRunnable 是同步执行源。没有调度操作符时，send 在订阅它的线程上执行；添加 subscribeOn 后，订阅和阻塞任务会调度到指定 Scheduler。

可以直接在 flatMap 的映射函数中判断线程，省去内部的 transformDeferred：

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

static void postProcess(Mono<Integer> output) {
    output
        .flatMap(message -> {
            Mono<Void> sending =
                Mono.fromRunnable(() -> send(message));

            // 仅适用于线程命名约定已被确认的环境。
            boolean alreadyOnBoundedElastic =
                Thread.currentThread()
                    .getName()
                    .startsWith("boundedElastic-");

            if (alreadyOnBoundedElastic) {
                return sending;
            }

            return sending.subscribeOn(Schedulers.boundedElastic());
        })
        .subscribe(
            ignored -> {},
            error -> System.err.println(
                "Unexpected error occurred during message sending: "
                    + error.getMessage()
            )
        );
}

这里只需修改 postProcess：

  • 线程判断为真时，内部 Mono 在当前线程执行 send。
  • 线程判断为假时，通过 subscribeOn 将 send 调度到 boundedElastic。

也可以保留原来的 flatMap + transformDeferred + subscribeOn。在给出的链路中，flatMap 处理上游值时会订阅内部 Mono，因此内部的 transformDeferred 能观察到预期线程。直接在映射函数中判断更容易理解。

线程名称不能提供严格保证

线程名称可以修改，自定义线程也能使用相同前缀。自定义 Scheduler、线程工厂或不同的运行配置，还可能采用其他命名方式。

按名称判断可能出现两种误判：

  • 已经在合适的线程上，却因为名称不同而再次调度。
  • 名称符合前缀，实际线程却不适合执行阻塞任务，导致错误地在当前线程执行。

Reactor 没有通用的公共 API 可以直接判断当前线程是否属于某个 boundedElastic Scheduler 实例。

Schedulers.isInNonBlockingThread() 也不能替代这个判断。它检查当前线程是否被 Reactor 识别为非阻塞线程。返回 false 并不表示当前线程属于 boundedElastic，普通的 main 线程也可能返回 false。

如果必须严格满足执行线程的要求，又无法可靠识别线程来源，建议始终调度这个阻塞任务：

static void postProcess(Mono<Integer> output) {
    output
        .flatMap(message ->
            Mono.<Void>fromRunnable(() -> send(message))
                .subscribeOn(Schedulers.boundedElastic())
        )
        .subscribe(
            ignored -> {},
            error -> System.err.println(
                "Unexpected error occurred during message sending: "
                    + error.getMessage()
            )
        );
}

这样仍有调度开销，但执行位置不再依赖线程名称。若确实需要避免额外调度，应用就需要提供可靠的线程或执行上下文标记，并明确保证这些标记的含义。

使用时还要注意什么

Mono.fromRunnable 完成后只发送完成信号，不发送原来的 message。如果后续还需要这个值，可以写成：

return sending
    .subscribeOn(Schedulers.boundedElastic())
    .thenReturn(message);

send 抛出的异常会进入响应式链路的错误处理。上游为空时，flatMap 不会调用 send。内部订阅由 flatMap 管理，无须在映射函数中手动调用 subscribe()。

少一次调度可以减少调度和排队开销，但阻塞调用仍会占用当前 worker，直到发送结束。高负载下能否因此改善延迟，需要结合实际负载测量；示例日志只能说明这次执行使用了哪个线程。

备注：内容仅供参考。