为什么 Play 框架不关闭 Akka Stream？答案

【问题标题】：Why doesn't Play framework close the Akka Stream?为什么 Play 框架不关闭 Akka Stream？
【发布时间】：2017-09-30 10:19:08
【问题描述】：

actor 初始化一个连接到 websocket 的 Akka 流。这是通过使用可以向其发送消息的Source.actorRef 完成的，然后由webSocketClientFlow 处理并由Sink.foreach 使用。这可以在以下代码中看到（源自akka docs）：

class TestActor @Inject()(implicit ec: ExecutionContext) extends Actor with ActorLogging {

  final implicit val system: ActorSystem = ActorSystem()
  final implicit val materializer: ActorMaterializer = ActorMaterializer()

  def receive = {
    case _ =>
  }

  // Consume the incoming messages from the websocket.
  val incoming: Sink[Message, Future[Done]] =
  Sink.foreach[Message] {
    case message: TextMessage.Strict =>
      println(message.text)
    case misc => println(misc)
  }

  // Source through which we can send messages to the websocket.
  val outgoing: Source[TextMessage, ActorRef] =
  Source.actorRef[TextMessage.Strict](bufferSize = 10, OverflowStrategy.fail)

  // flow to use (note: not re-usable!)
  val webSocketFlow = Http().webSocketClientFlow(WebSocketRequest("wss://ws-feed.gdax.com"))

  // Materialized the stream
  val ((ws,upgradeResponse), closed) =
  outgoing
    .viaMat(webSocketFlow)(Keep.both)
    .toMat(incoming)(Keep.both) // also keep the Future[Done]
    .run()

  // Check whether the server has accepted the websocket request.
  val connected = upgradeResponse.flatMap { upgrade =>
    if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
      Future.successful(Done)
    } else {
      throw new RuntimeException(s"Failed: ${upgrade.response.status}")
    }
  }

  // When the connection has been established.
  connected.onComplete(println)

  // When the stream has closed
  closed.onComplete {
    case Success(_) => println("Test Websocket closed gracefully")
    case Failure(e) => log.error("Test Websocket closed with an error\n", e)
  }

}

当播放框架重新编译时，它会关闭 TestActor 但不会关闭 Akka 流。只有当 websocket 超时时，流才会关闭。

这是否意味着我需要手动关闭流，例如，在 TestActor PostStop 函数中发送使用 Source.actorRef 创建的演员 PoisonPill？

注意：我也尝试注入 Materializer 和 Actorsystem 即：

@Inject()(implicit ec: ExecutionContext, implicit val mat: Materializer, implicit val system: ActorSystem)

Play重新编译时，流被关闭，但也会产生错误：

[error] a.a.ActorSystemImpl - Websocket handler failed with
Processor actor [Actor[akka://application/user/StreamSupervisor-62/flow-0-0-ignoreSink#989719582]] 
terminated abruptly

【问题讨论】：

标签： scala playframework akka akka-http

【解决方案1】：

在您的第一个示例中，您将在您的演员中创建一个演员系统。你不应该这样做——actor系统很昂贵，创建一个意味着启动线程池、启动调度程序等。另外，你永远不会关闭它，这意味着你遇到的问题比流不关闭要大得多- 你有资源泄漏，actor系统创建的线程池永远不会关闭。

所以本质上，每次您收到 WebSocket 连接时，您都在创建一个带有一组新线程池的新参与者系统，并且您永远不会关闭它们。在生产环境中，即使负载很小（每秒几个请求），您的应用程序也会在几分钟内耗尽内存。

一般来说，在 Play 中，您永远不应该创建自己的 Actor 系统，而应该注入一个。从演员内部，您甚至不需要注入它，因为它会自动注入 - context.system 让您可以访问创建演员的演员系统。与物化器类似，它们没有那么重，但如果你为每个连接创建一个，如果你不关闭它，你也可能会耗尽内存，所以你应该注入它。

所以当你注入它时，你会得到一个错误 - 这很难避免，尽管并非不可能。困难在于 Akka 本身并不能真正自动知道需要关闭什么顺序才能优雅地关闭事物，它应该先关闭你的演员，以便它可以优雅地关闭流，还是应该关闭流关闭，以便他们可以通知您的演员他们已关闭并做出相应的响应？

Akka 2.5 对此有一个解决方案，即托管关闭序列，您可以在其中注册要关闭的东西，然后 Actor 系统开始以某种随机顺序杀死东西：

https://doc.akka.io/docs/akka/2.5/scala/actors.html#coordinated-shutdown

您可以将它与 Akka 流 kill switches 结合使用，在应用程序的其余部分关闭之前优雅地关闭您的流。

但一般来说，关机错误是相当良性的，所以如果是我，我不会担心它们。

【讨论】：