Reactor Mono 发布到多个方法
Reactor Mono publish to multiple methods
我在将对象发布到多个方法时遇到问题。下面给出了我的代码的简化版本。
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();
您的 Mono
组中有两种模式:
aMono
是一个常量,由于直接变量赋值(你调用 getA()
一次)
- 另一方面,其他 monos 从运算符内部调用
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(...);
这就是您获得报告的日志数量和顺序的原因。
我在将对象发布到多个方法时遇到问题。下面给出了我的代码的简化版本。
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();
您的 Mono
组中有两种模式:
aMono
是一个常量,由于直接变量赋值(你调用getA()
一次)- 另一方面,其他 monos 从运算符内部调用
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(...);
这就是您获得报告的日志数量和顺序的原因。