对流的进出口使用单个参与者

Question

我开始探索akka-stream API，并希望对其进行测试。我有一个特定的用例，我似乎找不到合适的流式API来使用。

这个想法与以下内容相似，但有一点扭曲：

资料来源（1至10）。mapAsync（num）=

在这种情况下，ask将在未来返回一条消息。我想为发送给参与者的每个请求返回多条消息。例如：

1. ActorRef被发送请求（2）
2. ActorRef从流中发送N条消息。
3. ActorRef被发送请求（3）
4. ActorRef从流中发送M条消息。

我可以向Actor中添加额外的逻辑，以将这些额外的消息聚合/缓冲到某种类型的Iterable中，但我想知道API中是否缺少处理此类情况的内容。

在查看文档时，我最接近的方法是使用水槽。actorRefWithAck和Source。排队等候。从SinkandSourceMat，并在确认OnInit消息之前将具体化的SourceQueue传递给actorRef。这允许在通过流量处理一对多消息的同时控制上下游背压。让图表更容易理解似乎适得其反。

Answer 1

不幸的是，你正在寻找的似乎并不存在。ask模式，例如 方法，只能返回1个值。因此，当你做类似

type Input = ???
type Outupt = ???

def askActorFlow(actorRef : ActorRef) : Flow[Input, Output, NotUsed] = 
  Flow[Input] mapAsync { input => (actorRef ? input).mapTo[Output] }

一次只能从ActorRef获取1个值。

您建议的工作，例如“在Actor中添加额外的逻辑以将这些额外的消息聚合/缓冲到Iterable中”，无论如何都是一个更好的解决方案。它使您的Actor用户更容易知道他们对任何查询只会得到一个响应。否则，用户将永远无法真正知道他们是否在查询后从Actor那里得到了“完整”的答案，或者是否会收到更多数据。

Answer 2

我相信您正在寻找flatMapConcat。

Source(1 to 10).flatMapConcat(num => 
  // Dummy example - outputs first 5 successors for each incoming element
  Source(num to (num + 5))
).to(Sink.foreach(println))

通过这种方式，你可以让Akka提升所有反应流的重量，并且不让任何一个赤裸裸的参与者污染你的管道。

这里有文档。