Akka Streams 中的 Flow#join 是什么答案

【问题标题】：What is Flow#join in Akka StreamsAkka Streams 中的 Flow#join 是什么
【发布时间】：2016-03-07 01:42:37
【问题描述】：

我是 Scala 中 Akka Streams 的学习者。当我在IncomingConnection 阅读时，我发现了Flow#join。然后，我在Flow#join的评论上找到了下面这张图。

+------+        +-------+
|      | ~Out~> |       |
| this |        | other |
|      | <~In~  |       |
+------+        +-------+

但是，我想知道它的结构是什么。我认为“加入”会形成一个循环。

所以我想让你解释一下“join”的结构，并告诉我使用Flow#join的简单示例代码

【问题讨论】：

标签： scala akka akka-stream

【解决方案1】：

文档状态：

通过交叉连接输入和输出，将此流加入另一个流，创建一个RunnableGraph

这是来自akka.http.scaladsl 的一个很好的例子，可以帮助解释为什么这是有用的：

  /**
   * Represents one accepted incoming HTTP connection.
   */
  final case class IncomingConnection(
    localAddress: InetSocketAddress,
    remoteAddress: InetSocketAddress,
    flow: Flow[HttpResponse, HttpRequest, NotUsed]) {

    /**
     * Handles the connection with the given flow, which is materialized exactly once
     * and the respective materialization result returned.
     */
    def handleWith[Mat](handler: Flow[HttpRequest, HttpResponse, Mat])(implicit fm: Materializer): Mat =
      flow.joinMat(handler)(Keep.right).run()

您可能知道 Akka http 流的 handler 始终从 HttpRequest 流向 HttpResponse，但正如您所见，IncomingConnection.flow 从 HttpResponse 流向 HttpRequest。换句话说，用户有责任从请求中创建响应，而 Akka Http 有责任发送该响应并产生另一个请求。当涉及到另一个 Flow 时，这确实会形成一个闭环，因此 join 方法会创建一个 RunnableGraph。

要了解如何处理连接，您应该了解更多关于BidiFlow 的信息。 BidiFlow#join 的结果是另一个流，因为 BidiFlow 有两个输入和两个输出。这是带有示例的link to an excellent explanation。

【讨论】：

非常感谢您的回复！你的回答对我很有帮助，因为我想了解更多IncommingConnection.handlerWith。没注意到IncommingConnection.flow的类型是Flow[HttpResponse, HttpRequest]：我误会是Flow[HttpRequest, HttpResponse]。我访问了链接，阅读并发现了有关 Akka Streams 的新发现。我学到了更多，谢谢。我尝试运行示例日志记录代码。但不幸的是，它不起作用。可以正常编译，但是没有输出。我尝试了很多我知道的方法，但我做不到。
没问题。也许您可以问另一个问题，为什么您编译的日志记录示例似乎不起作用。如果您能同时接受我的回答，我将不胜感激。