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

Question

有没有办法确保从未来创建的单个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，因为未来是从"完成器"线程完成的。我正在尝试弄清楚是否有办法使其始终打印Thread: main.

Answer 1

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

您可以做的最接近的事情是确保：

1. 订阅在特定Scheduler(通过.subscribeOn)执行，并且
2. 未来的完成值在同一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并不强制要求所有运营商都使用该Scheduler进行操作。 它只影响下游运算符，直到下一个.publishOn，或者直到下一个异步运算符(如.flatMap)可能使用不同的Scheduler。

Answer 2

作为一个非常非优化的概念证明，这可以通过以下方式实现：

让我们创建一个能够以受控方式"按需"执行任务的执行器。

private static class SelfEventLoopExecutor implements Executor {
private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
@Override
public void execute(Runnable command) {
boolean added = queue.add(command);
assert added;
}
public void drainQueue() {
Runnable r;
while ((r = queue.poll()) != null) {
r.run();
}
}
}

接下来，创建一个订阅者，该订阅器能够在等待结果的同时使用执行器执行任务，而不是完全阻塞线程。

public static class LazyBlockingSubscriber<T> implements Subscriber<T> {
private final SelfEventLoopExecutor selfExec;
private volatile boolean completed = false;
private volatile T value;
private volatile Throwable ex;
public LazyBlockingSubscriber(SelfEventLoopExecutor selfExec) {
this.selfExec = selfExec;
}
@Override
public void onSubscribe(Subscription s) {
s.request(1);
}
@Override
public void onNext(T t) {
value = t;
completed = true;
}
@Override
public void onError(Throwable t) {
ex = t;
completed = true;
}
@Override
public void onComplete() {
completed = true;
}
public T block() throws Throwable {
while (!completed) {
selfExec.drainQueue();
}
if (ex != null) {
throw ex;
}
return value;
}
}

现在，我们可以通过以下方式修改代码

public static void main(String[] args) throws Throwable {
var future = new CompletableFuture<String>();
var selfExec = new SelfEventLoopExecutor(); // our new executor
var res = Mono.fromFuture(future)
.publishOn(Schedulers.fromExecutor(selfExec))  // schedule on the new executor
.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();
var subs = new LazyBlockingSubscriber<String>(selfExec); // lazy subscribe
res.subscribeWith(subs);
subs.block(); // spin wait
}

因此，代码打印Thread: main.