java发布到多个方法

Question

我在将对象发布到多个方法时遇到问题。下面给出了我的代码的简化版本

package org.example.reactive;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple3;

@Slf4j
public class MonoTest {

    public static void main(String... args) {
        MonoTest m = new MonoTest();
        Mono<A> aMono = m.getA();
        Mono<B> bMono = aMono.flatMap(m::getB);
        Mono<C> cMono = aMono.flatMap(m::getC);
        Mono<D> dMono = cMono.zipWith(bMono).flatMap(m::getD);
        Mono<E> eMono = Mono.zip(aMono, cMono, dMono)
                .flatMap(m::getE);
        aMono
                .zipWith(eMono)
                .subscribe(m::onCompleted, m::onFailed);
    }

    private Mono<A> getA(){
        log.info("inside getA");
        return Mono.just(new A());
    }
    private Mono<B> getB(A a){
        log.info("inside getB");
        return Mono.just(new B());
    }
    private Mono<C> getC(A a){
        log.info("inside getC");
        return Mono.just(new C());
    }
    private Mono<D> getD(Tuple2 t){
        log.info("inside getD");
        return Mono.just(new D());
    }
    private Mono<E> getE(Tuple3 t){
        log.info("inside getE");
        return Mono.just(new E());
    }
    private void onCompleted(Tuple2 t){
        log.info("inside onCompleted");
    }
    private void onFailed(Throwable t){
        log.info("inside onFailed");
    }

    class A {}
    class B {}
    class C {}
    class D {}
    class E {}
}

我希望每个方法只调用一次。但是getC被调用两次。这里怎么了？程序输出如下

组织。实例反应性。MonoTest-内部getA

组织。实例反应性。MonoTest-内部getC

组织。实例反应性。MonoTest-内部getC

组织。实例反应性。MonoTest-内部getB

组织。实例反应性。MonoTest-内部getD

组织。实例反应性。MonoTest-内部getE

组织。实例反应性。单一测试-内部未完成

编辑

我可以通过如下缓存来解决它

        Mono<A> aMono = m.getA().cache();
        Mono<B> bMono = aMono.flatMap(m::getB).cache();
        Mono<C> cMono = aMono.flatMap(m::getC).cache();
        Mono<D> dMono = cMono.zipWith(bMono).flatMap(m::getD).cache();

Answer 1

在Mono集合中有两种模式：

• aMono是一个常量，由于直接变量分配（您调用getA()一次），所以需要立即解析一次
• 另一方面，其他monos从操作符内部调用getX()方法，尤其是flatMap。这意味着当订阅flatmapping mono时，这些调用会被延迟执行

aMono是getX()方法的唯一顶级调用。 用它们的定义替换Mono变量，除了aMono之外，应该会更清楚发生什么：

MonoTest m = new MonoTest();

Mono<A> aMono = m.getA(); // <  getA log
aMono.zipWith(
    Mono.zip(
       aMono,
       aMono.flatMap(m::getC),  // <  getC log
       aMono.flatMap(m::getC) // <  getC log
            .zipWith(aMono.flatMap(m::getB)) // <  getB log
            .flatMap(m::getD) // <  getD log
    ).flatMap(m::getE) // <  getE log
  ).subscribe(...);

这就是为什么您会得到报告日志的数量和顺序