如何使用Akka组合Source.repeak和Source.comcompletionStage

Question

我使用的是带有微服务框架的akka，所以我收到了很多completionStage请求。我想从一个微服务中获得一个元素列表，并将它们与另一微服务中的单个元素压缩在一起，这样我就可以得到一个源对<列表项，单个元素>。

我不能用普通的zip来做这件事，因为只要两个源中的一个完成，Source.zip就会完成，所以我最终只传播了一个元素。

我不能使用Source.zip All，因为这需要我提前定义默认元素。

如果我已经提前拥有了单个元素，我可以使用Source.repeak使其重复传播该元素，这意味着Source.zip将在元素列表完成时完成，但Source.repeac不能处于完成阶段或Source.commpletionStage.

我目前的策略是在映射Concat列表元素之前将这些东西压缩在一起。

Source<singleElement> singleElement = Source.completionStage(oneService.getSingleElement().invoke());
return Source.completionStage(anotherService.getListOfElements().invoke)
.zip(singleElement)
.flatMapConcat(pair -> Source.fromIterator(() -> pair.first().stream().map(listElement -> Pair.create(listElement, pair.second())));

这最终达到了我想要的效果，但我觉得有很多不必要的重复和数据的同步移动。有没有更好的方法来解决我错过的这个问题？

Answer 1

flatMapConcat运算符应该允许您构造一个Source.repeat，一旦知道单个元素，它就会重复该元素。在Scala中(Source.future相当于Source.completionStage的Scala：我对Java lambda语法不够熟悉，无法用Java回答)：

val singleElement = Source.future(oneService.getSingleElement)
Source.future(anotherService.getListOfElements)
.mapConcat(lst => lst)  // unspool the list
.zip(singleElement.flatMapConcat(element => Source.repeat(element)))

Answer 2

为什么不合并CompletionStage，然后将它们输入Akka流？

Source<Pair<String,String>, ?> execute() {
CompletionStage<Pair<String, List<String>>> pairCompletionStage = getSingleElement().thenCombine(getListOfElements(), Pair::create);
return Source.completionStage(pairCompletionStage)
.flatMapConcat(pair -> Source.from(pair.second()).map(listElement -> Pair.create(listElement, pair.first())));
}

在睡眠超时的情况下进行完整的PoC播放，以首先完成一个或另一个CompletionStage：

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import akka.Done;
import akka.actor.ActorSystem;
import akka.japi.Pair;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
public class CompletionStages {
CompletionStage<String> getSingleElement() {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
return "Single Element";
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
});
}
CompletionStage<List<String>> getListOfElements() {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(3000);
return Arrays.asList("One", "Two", "Three");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
});
}
Source<Pair<String,String>, ?> execute() {
CompletionStage<Pair<String, List<String>>> pairCompletionStage = getSingleElement().thenCombine(getListOfElements(), Pair::create);
return Source.completionStage(pairCompletionStage)
.flatMapConcat(pair -> Source.from(pair.second()).map(listElement -> Pair.create(listElement, pair.first())));
}
CompletionStage<Done> run(ActorSystem system) {
return execute().runWith(Sink.foreach(System.out::println), system);
}
public static void main(String... args) {
final ActorSystem system = ActorSystem.create();
new CompletionStages().run(system)
.thenRun(system::terminate);
}
}