Reactor Mono发布到多个方法

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

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 被调用了两次。这里有什么问题呢?程序输出如下

org.example.reactive.MonoTest – 在getA内。

org.example.reactive.MonoTest – 在getC里面

org.example.reactive.MonoTest – 在getC里面

org.example.reactive.MonoTest – 在getB里面。

org.example.reactive.MonoTest – 在getD内。

org.example.reactive.MonoTest – 在getE里面。

org.example.reactive.MonoTest – 在onCompleted里面。

编辑

好吧,我可以通过如下的缓存来解决。

        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();

解决方案:

在你的这组代码中有两种模式 Monos:

  • aMono 是一个常数,并且由于直接的变量同化而被急切地解析了一次(你调用了 getA() 一次)
  • 另一方面,其他单体称 getX() 的方法,特别是 flatMap. 这意味着当平移单体被认购时,这些调用会被懒惰地执行。

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(...);

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

给TA打赏
共{{data.count}}人
人已打赏
未分类

当我在新电脑上用Visual Studio运行ASP.NET MVC时,我向它提出的每一个请求都会自动重定向到微软的登录。

2022-9-8 22:55:22

未分类

异常在构造函数try块中被捕获,并被处理,但仍然会被第二次重抛。

2022-9-8 22:55:24

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索