java - 在 Mono 的阻塞线程上执行转换步骤

Question

有没有办法确保从 future 创建的单个 Mono 的所有转换步骤都在订阅和阻塞的线程上执行？

例如下面的代码

public static void main(String[] args) {
  var future = new CompletableFuture<String>();
  var res = Mono.fromFuture(future).map(val -> {
    System.out.println("Thread: " + Thread.currentThread().getName());
    return val + "1";
  });

  new Thread(() -> {
    try {
      Thread.sleep(1000L);
    } catch (InterruptedException e) {
    }
    future.complete("completed");
  }, "completer").start();

  res.block();
}

打印 Thread:completer 因为 future 是从“completer”线程完成的。我试图弄清楚是否有办法让它始终打印 Thread: main。

Answer 1

没有。当 main 线程通过 .block() 被阻塞时，该线程专门等待 onNext、onComplete ，或流的 onError 信号(在所有上游运算符执行完毕之后)。在调用上游运算符以执行运算符之前，它不会以某种方式重新获得控制权。

您能做的最接近的事情是确保:

1. 订阅在特定的Scheduler上执行(通过.subscribeOn)并且
2. future 的完成值在同一个 Scheduler 上发布(通过 .publishOn)。

例如:

Scheduler scheduler = Schedulers.parallel();
var res = Mono.fromFuture(future)
        .doFirst(() -> {   // Note: doFirst added in 3.2.10.RELEASE
            // prints a thread in the parallel Scheduler (specified by subscribeOn below)
            System.out.println("Subscribe Thread: " + Thread.currentThread().getName());
        })
        // specifies the Scheduler on which the the completion value
        // from above is published for downstream operators
        .publishOn(scheduler)
        .map(val -> {
            // prints a thread in the parallel Scheduler (specified by publishOn above)
            System.out.println("Operator Thread: " + Thread.currentThread().getName()); 
            return val + "1";
        })
        // specifies the Scheduler on which  upstream operators are subscribed
        .subscribeOn(scheduler);

但是，请注意以下几点:

• 订阅发生在Scheduler 中的线程上，而不是发生在阻塞的main 线程上。
• 此方法只是确保在 Scheduler 中使用相同的 Scheduler，而不是相同的 Thread。理论上，您可以使用单线程调度程序强制使用相同的Thread(例如Schedulers.newParallel("single-threaded", 1))
• .publishOn 不会强制所有操作符都在该调度程序上运行。它只会影响下游运算符，直到下一个 .publishOn，或者直到下一个可能使用不同 Scheduler 的异步运算符(例如 .flatMap) .