在 Project Reactor 中编写命令式代码

Question

我有一个类有以下两种方法。

public class Test1 {
    public Mono<String> blah1() {
        Mono<String> blah = Mono.just("blah1");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("blah1 done");
        return blah;
    }
    
    public Mono<String> blah2() {
        Mono<String> blah = Mono.just("blah2");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("blah2 done");
        return blah;
    }
}

我有以下 JUnit：

@Test
public void blah1Test() {
    Flux<Tuple2<String, String>> s = Flux.zip(test1.blah1(), test1.blah2());    
}

我的结果如下：

blah1 done
blah2 done

我希望 blah2 在 blah1 之前完成。因此我相信这是处理阻塞而不是非阻塞。我需要做什么才能将输出切换到 blah2 完成然后 blah1 完成？基本上为什么这些不并行处理？

提前感谢您的宝贵时间！

Answer 1

sleep 和 println 都在反应式管道之外执行。因此，blah1() 和 blah2() 的行为都类似于常规的非反应性方法。

试试这个：

public Mono<String> blah1() {
    System.out.println("blah1 start");
    return Mono.just("blah1")
        .delayElement(Duration.ofMillis(5000))
        .doOnNext(e -> System.out.println("blah1 done"));
}

public Mono<String> blah2() {
    System.out.println("blah2 start");
    return Mono.just("blah2")
        .delayElement(Duration.ofMillis(1000))
        .doOnNext(e -> System.out.println("blah2 done"));
}

在这里，我们得到了预期的结果，因为打印发生在响应式管道中。

输出：

blah1  start 
blah2 start 
blah2 done
blah1 done