我想创建一个结合了两个反应性来源的结果的服务。一个正在生产Mono，另一个正在生产Flux。对于合并，我需要为每个发射的通量使用相同的mono值。
现在我有这样的东西
Flux.zip(
service1.getConfig(), //produces flux
service2.getContext() //produces mono
.cache().repeat()
)
这给了我我需要的东西，
但是我已经注意到，上下文被缓存后，repeat（）发出了大量的元素。这有问题吗？
是否可以将重复次数限制为接收到的配置数量，但是仍然同时请求两个？还是这不是问题，我可以安全地忽略那些额外发出的元素吗？
我尝试使用combineLatest，但根据时间安排，某些配置元素可能会丢失并且无法处理。
编辑
参考@Ricard Kollcaku的建议，我创建了示例测试，以显示为什么这不是我想要的。
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
public class SampleTest
{
Logger LOG = LoggerFactory.getLogger(SampleTest.class);
AtomicLong counter = new AtomicLong(0);
Flux<String> getFlux()
{
return Flux.fromStream(() -> {
LOG.info("flux started");
sleep(1000);
return Stream.of("a", "b", "c");
}).subscribeOn(Schedulers.parallel());
}
Mono<String> getMono()
{
return Mono.defer(() -> {
counter.incrementAndGet();
LOG.info("mono started");
sleep(1000);
return Mono.just("mono");
}).subscribeOn(Schedulers.parallel());
}
private void sleep(final long milis)
{
try
{
Thread.sleep(milis);
}
catch (final InterruptedException e)
{
e.printStackTrace();
}
}
@Test
void test0()
{
final Flux<String> result = Flux.zip(
getFlux(),
getMono().cache().repeat()
.doOnNext(n -> LOG.warn("signal on mono", n)),
(s1, s2) -> s1 + " " + s2
);
assertResults(result);
}
@Test
void test1()
{
final Flux<String> result =
getFlux().flatMap(s -> Mono.zip(Mono.just(s), getMono(),
(s1, s2) -> s1 + " " + s2));
assertResults(result);
}
@Test
void test2()
{
final Flux<String> result = getFlux().flatMap(s -> getMono().map((s1 -> s + " " + s1)));
assertResults(result);
}
void assertResults(final Flux<String> result)
{
final Flux<String> flux = result;
StepVerifier.create(flux)
.expectNext("a mono")
.expectNext("b mono")
.expectNext("c mono")
.verifyComplete();
Assertions.assertEquals(1L, counter.get());
}
查看test1和test2的测试结果
2020-01-20 12:55:22.542 INFO [] [] [ parallel-3] SampleTest : flux started
2020-01-20 12:55:24.547 INFO [] [] [ parallel-4] SampleTest : mono started
2020-01-20 12:55:24.547 INFO [] [] [ parallel-5] SampleTest : mono started
2020-01-20 12:55:24.548 INFO [] [] [ parallel-6] SampleTest : mono started
expected: <1> but was: <3>
我需要拒绝你的建议。在这两种情况下，getMono都是-调用与flux中的项目一样多的次数-在通量的第一个元素到达后调用这些是我要避免的互动。我的服务正在发出HTTP请求，这可能很耗时。
我当前的解决方案没有这个问题，但是如果我将记录器添加到我的邮政编码中，我会得到这个
2020-01-20 12:55:20.505 INFO [] [] [ parallel-1] SampleTest : flux started
2020-01-20 12:55:20.508 INFO [] [] [ parallel-2] SampleTest : mono started
2020-01-20 12:55:21.523 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.528 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.529 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.529 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.529 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.529 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.530 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.530 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.530 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.530 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.531 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.531 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.531 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.531 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.531 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.532 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.532 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.532 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.532 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.533 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.534 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.534 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.534 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.534 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.534 WARN [] [] [ parallel-2] SampleTest : signal on mono
2020-01-20 12:55:21.535 WARN [] [] [ parallel-2] SampleTest : signal on mono
如您所见，将cache().repeat()组合在一起会发出很多元素，我想知道这是否是一个问题，如果是，那么如何避免它（但请保持单次调用为单调用和并行调用）。
您只需简单的更改就可以完成
getFlux()
.flatMap(s -> Mono.zip(Mono.just(s),getMono(), (s1, s2) -> s1+" "+s2))
.subscribe(System.out::println);
Flux<String> getFlux(){
return Flux.just("a","b","c");
}
Mono<String> getMono(){
return Mono.just("mono");
}
如果您想使用zip，但可以使用平面图实现相同的结果
getFlux()
.flatMap(s -> getMono()
.map((s1 -> s + " " + s1)))
.subscribe(System.out::println);
}
Flux<String> getFlux() {
return Flux.just("a", "b", "c");
}
Mono<String> getMono() {
return Mono.just("mono");
}
两者的结果都是：单声道b单声道c mono
像Project Reactor和RxJava之类的库试图提供尽可能多的功能组合，但是不提供对组合功能的工具的访问。结果，总有一些未涵盖的极端情况。
据我所知，我自己的DF4J是唯一提供组合功能的方法的异步库。例如，这是用户压缩Flux和Mono的方式：（当然，此类不是DF4J本身的一部分）：
import org.df4j.core.dataflow.Actor;
import org.df4j.core.port.InpFlow;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
abstract class ZipActor<T1, T2> extends Actor {
InpFlow<T1> inpFlow = new InpFlow<>(this);
InpFlow<T2> inpScalar = new InpFlow<>(this);
ZipActor(Flux<T1> flux, Mono<T2> mono) {
flux.subscribe(inpFlow);
mono.subscribe(inpScalar);
}
@Override
protected void runAction() throws Throwable {
if (inpFlow.isCompleted()) {
stop();
return;
}
T1 element1 = inpFlow.removeAndRequest();
T2 element2 = inpScalar.current();
runAction(element1, element2);
}
protected abstract void runAction(T1 element1, T2 element2);
}
这就是它的使用方式：
@Test
public void ZipActorTest() {
Flux<Integer> flux = Flux.just(1,2,3);
Mono<Integer> mono = Mono.just(5);
ZipActor<Integer, Integer> actor = new ZipActor<Integer, Integer>(flux, mono){
@Override
protected void runAction(Integer element1, Integer element2) {
System.out.println("got:"+element1+" and:"+element2);
}
};
actor.start();
actor.join();
}
控制台输出如下：
got:1 and:5
got:2 and:5
got:3 and:5