这个例子有点奇怪，但据我所知，您希望调用前两个服务，每个服务返回一个值。之后，如果需要，您希望调用第三个字段，并将该字段中的值设置为第一个字段中的一个字段。
无论如何，有一个简单的解决方案，但是有了更多的信息，也许我们可以创建更好的流。此流采用flatmap的adventages，后者急切地订阅内部发布服务器。
[这个例子是用kotlin编写的，非常像java。唯一让人困惑的是 it 变量，等于： map(it -> it.sg ) ]

data class Example(
    val name: String,
    val status: String,
    var value: String? = null  
)

class ReactorTest {

    @Test
    fun test() {
        val first = Mono.just(Example("first", "suspended"))
        val second = Mono.just(Example("second", "suspended"))
        val third = Mono.just(Example("third", "suspended", "thirdValue"))

        val stream = first.zipWith(second)
            .flatMap { tuple ->
                Mono.just(tuple.t1)
                    .filter { it.status == tuple.t2.status }
                    .zipWith(third)
                    .doOnNext {
                        it.t1.value = it.t2.value
                    }
                    .map { it.t1 }
                    .switchIfEmpty(Mono.just(tuple.t1))
            }

            StepVerifier.create(stream)
                .expectNext(Example("first", "suspended", "thirdValue"))
                .verifyComplete()
    }

    @Test
    fun test2() {
        val first = Mono.just(Example("first", "suspended"))
        val second = Mono.just(Example("second", "active"))
        val third = Mono.just(Example("third", "suspended", "thirdValue"))

        val stream = first.zipWith(second)
            .flatMap { tuple ->
                Mono.just(tuple.t1)
                    .filter { it.status == tuple.t2.status }
                    .zipWith(third)
                    .doOnNext {
                        it.t1.value = it.t2.value
                    }
                    .map { it.t1 }
                    .switchIfEmpty(Mono.just(tuple.t1))
            }

        StepVerifier.create(stream)
            .expectNext(Example("first", "suspended"))
            .verifyComplete()
    }
}

旁注：如果您在React流中使用阻塞服务，那么应该将这些服务分离到专用的线程池中。比如：

fun blockingService(): Mono<String> {
    //real service use fromCallable
    return Mono.just("fromCallableOnServiceCall")
        //for real service it may use a dedicated pool
        .subscribeOn(Schedulers.boundedElastic())

}