reactor Publisher中使用delayElement时如何使用主线程

Question

在reactor Publisher中使用delayElement时如何使用主线程。

Mono.just("one")
    .delayElement(Duration.ofSeconds(3))
    .subscribe(System.out::println);

如果我运行此代码，它不会打印“一”，因为主线程将退出并且订阅在其他线程上完成。

当我使用delayElement时，有没有办法使用主线程来订阅这个Mono？ 即等待 3 秒后发布“one”，并直到订阅者消费它。

请注意，如果我不使用“delayElement”，我可能会阻塞主线程，直到它在屏幕上打印“one”。

Mono.just("one")
    .subscribe(System.out::println);

我知道我可以通过使用 wait 和 notification （或类似的替代方法）来实现类似的输出，但我想要的是，在调用 delayElement 后使用主线程来订阅元素。

Mono 使用的类是从 reactor.core.publisher.Mono 导入的

考虑下面的示例，其中我将使用相同的线程来延迟操作自己的实现。

Mono.just("one")
    .doOnNext(a -> this.myDelayOperation(Duration.ofSeconds(1)))
    .subscribe(System.out::println);

其中 myDelayOperation(...) 是

private void myDelayOperation(Duration duration) {
    try {
        Thread.sleep(duration.toMillis());
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

Answer 1

免责声明：所有这些解决方案都不建议用于生产。仅用于测试和学习目的。

调试：在代码中添加日志

    public static void main(String[] args) {
            System.out.println("Method Start Thread : "+Thread.currentThread().getName());
            Mono.just("one")
                    .doOnNext(next->System.out.println("Before delay Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                    .delayElement(Duration.ofSeconds(5))
                    .doOnNext(next->System.out.println("After delay Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                    .subscribe(val->System.out.println("Final Value: "+val+" Value Thread : "+Thread.currentThread().getName()));
            System.out.println("Method Exit Thread: "+Thread.currentThread().getName());
    }

输出：

• 方法启动线程：main
• 延迟 Value:one 和 Value Thread:main 之前
• 方法退出线程：main
• 进程已完成，退出代码为 0

说明：这里注意，在打印“延迟值之后”之前，主线程已退出。因为delayElement()不会在主线程上运行，而是在称为“并行”的不同线程上运行。在并行线程完成其操作之前，主线程已退出，因此您看不到值“one”

解决方案1：使主线程在一段时间内不退出

public static void main(String[] args) {
        System.out.println("Method Start Thread : "+Thread.currentThread().getName());
        Mono.just("one")
                .doOnNext(next->System.out.println("Before delay Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                .delayElement(Duration.ofSeconds(5))
                .doOnNext(next->System.out.println("After delay Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                .subscribe(val->System.out.println("Final Value: "+val+" Value Thread : "+Thread.currentThread().getName()));

        try {
            System.out.println("Sleeping Thread: "+Thread.currentThread().getName());
            Thread.sleep(10000);
            System.out.println("Wake up Thread: "+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("Method Exit Thread: "+Thread.currentThread().getName());

输出：

• 方法启动线程：main
• 延迟 Value:one 和 Value Thread: main 之前
• 睡眠线程：主
• 延迟值：one 和值线程：parallel-1 后
• 最终值：一个值线程：parallel-1
• 唤醒线程：main
• 方法退出线程：main
• 进程已完成，退出代码为 0

说明：这里请注意，“延迟值之后”是在主线程处于睡眠状态时（在退出之前）打印的。另请注意，如前所述，delayElement() 不在主线程上运行，而是在称为“parallel-1”的不同线程上运行。

解决方案：2 使用 block()

public static void main(String[] args) {
        System.out.println("Method Start Thread : "+Thread.currentThread().getName());
        Mono.just("one")
                .doOnNext(next->System.out.println("Before delay Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                .delayElement(Duration.ofSeconds(5))
                .doOnNext(next->System.out.println("After delay Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                .block();
        System.out.println("Method Exit Thread: "+Thread.currentThread().getName());

输出：

• 方法启动线程：main
• 延迟 Value:one 和 Value Thread: main 之前
• 延迟值：one 和值线程：parallel-1 后
• 方法退出线程：main
• 进程已完成，退出代码为 0

说明： 这里请注意，因为我们阻塞了主线程，所以打印了字符串“延迟值之后”。因此，在管道返回值之前，它将被阻塞。

解决方案 3：编写自定义执行器：强烈不推荐

.delayElement(Duration.ofSeconds(5), Schedulers.fromExceutor(wirteAndCallYourOwnExecutorToBlockCurrentMainThread))

解决方案 4：使用 Mono Callable 进行阻止

public static void main(String[] args) {
        System.out.println("Method Start Thread : "+Thread.currentThread().getName());
        Mono.fromCallable(()->{
                    try {
                        System.out.println("Sleeping Thread: "+Thread.currentThread().getName());
                        Thread.sleep(10000);
                        System.out.println("Wake up Thread: "+Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return "one";
                })
                .doOnNext(next->System.out.println("Received Value:"+next+" and Value Thread: "+Thread.currentThread().getName()))
                .subscribe();
        System.out.println("Method Exit Thread: "+Thread.currentThread().getName());

输出：

• 方法启动线程：main
• 睡眠线程：主
• 唤醒线程：main
• 收到的值：one 和值线程：main
• 方法退出线程：main
• 进程已完成，退出代码为 0

>建议：如果是产品，最好使用异步而不阻塞。我不知道你的完整用例。因此，在最坏的情况下，如果需要，请使用方法 4，风险自负。