如何从主要方法终止 Akka 演员系统？答案

【问题标题】：How to terminate Akka actor system from a main method?如何从主要方法终止 Akka 演员系统？
【发布时间】：2017-05-29 02:27:59
【问题描述】：

我有一个使用 Akka Streams 和 ReactiveMongo 的 this 应用程序。没有用户定义的演员。应用程序从main 方法启动。

问题是 JVM 在 main 方法完成后继续永远运行。这就是我现在正在做的事情：

val g = (file: String) => RunnableGraph.fromGraph(GraphDSL.create(Sink.ignore) {
  implicit builder =>
    sink =>
      import GraphDSL.Implicits._

      // Source
      val A: Outlet[(String, String)] = builder.add(Source.fromIterator(() => parseMovies(file).iterator)).out
      // Flow
      val B: FlowShape[(String, String), Either[String, Movie]] = builder.add(findMovie)
      // Flow
      val C: FlowShape[Either[String, Movie], Option[String]] = builder.add(persistMovie)

      A ~> B ~> C ~> sink.in

      ClosedShape
})

def main(args: Array[String]): Unit = {
  require(args.size >= 1, "Path to file is required.")

  g(args(0)).run
    .onComplete(_ => Await.result(system.terminate(), 5.seconds))
}

我已经阅读了this 线程和this，它们都不起作用。 system.shutdown 已弃用，我没有任何明确的演员要注意。我可以打电话给system.exit，但这并不优雅。

从日志看来，Akka 正在尝试关闭，但随后我看到一堆 Mongo 消息。

2017-01-13 11:35:57.320 [DEBUG] a.e.EventStream.$anonfun$applyOrElse$4 - shutting down: StandardOutLogger started
2017-01-13 11:36:05.397 [DEBUG] r.c.a.MongoDBSystem.debug - [Supervisor-1/Connection-2] ConnectAll Job running... Status: {{NodeSet None Node[localhost:27017: Primary (10/10 available connections), latency=6], auth=Set() }}
2017-01-13 11:36:05.420 [DEBUG] r.c.a.MongoDBSystem.debug - [Supervisor-1/Connection-2] RefreshAll Job running... Status: {{NodeSet None Node[localhost:27017: Primary (10/10 available connections), latency=6], auth=Set() }}
// more of MongoDBSystem.debug messages

为什么不会it.just.die？

【问题讨论】：

你应该正确关闭MongoDriver，因为它管理着自己的底层Actor系统，你不应该（不能）直接访问
@cchantep 你能说得更具体些或显示一个代码示例吗？关闭 Mongo 连接没有帮助。
如果你仔细阅读我提到了MongoDriver，而不是MongoConnection
@cchantep 我第一次理解你，但要求提供代码示例。无论如何，我做了以下更改，它确实有效。如果您发布答案（而不是评论），我会接受。 g(args(0)).run .onComplete(_ => { driver.close(5.seconds) Await.result(system.terminate(), 5.seconds) })

标签： scala akka akka-stream akka-http reactivemongo

【解决方案1】：

我想你想添加一个关闭挂钩或致电actorSystem.registerOnTermination(driver.close())：

def main(args: Array[String]): Unit = {
  import akka.actor.CoordinatedShutdown
  require(args.size >= 1, "Path to file is required.")

  CoordinatedShutdown(system).addTask(CooordinatedShutdown.PhaseBeforeActorSystemTerminate, "shutdownMongoDriver") { () => driver.close(5.seconds); Future.successful(Done) }

  g(args(0)).run.onComplete(_ => CoordinatedShutdown(system).run())    
}

【讨论】：