使用 Akka Streams 设置正常关闭的更好方法

Question

我有一个非常简单的应用程序，它有一个 Akka HTTP 端点，进行一些处理并将结果写入任一输出文件。确保优雅关闭的代码看起来有点复杂，有没有办法让它更简洁？

  val bindingFuture = Http().newServerAt("localhost", config.port).bind(route)

  val validQueue: BoundedSourceQueue[ByteString] = ???
  val invalidQueue: BoundedSourceQueue[ByteString] = ???
  val validDone: Future[Done] = ???
  val invalidDone: Future[Done] = ???
  val allDone = Future.sequence(validDone, invalidDone)

  bindingFuture.onComplete {
    case Success(binding) =>
      logger.info("Server started on port {}", config.port)
      binding.addToCoordinatedShutdown(5.seconds)
    case Failure(ex) =>
      logger.error("Can't start server", ex)
      system.terminate()
  }

  allDone.onComplete { result =>
    result match {
      case Failure(ex) =>
        logger.error("Streams completed with error", ex)
      case Success(_) =>
        logger.info("Streams completed successfully")
    }
    system.terminate()
  }

  sys.addShutdownHook {
    logger.info("Shutting down...")
    validQueue.complete()
    invalidQueue.complete()
  }

Answer 1

Akka默认安装JVM关闭钩子，不需要自己添加关闭钩子，可以去掉sys.addShutdownHook { ... } 调用ActorSystem.terminate 也将终止所有流。 （流可以突然终止，但在大多数应用程序中这不是问题，听起来这在您的情况下也不是问题。）

轻微清理，您可以考虑使用map、recoverWith 和andThen。

allDone.map { _ =>
  logger.info("Streams completed successfully")
}.recoverWith {
  case ex => 
    logger.error("Streams completed with error", ex)
}.andThen {
  case _ => system.terminate()
}

您可以使用 CoordinatedShutdown：

CoordinatedShutdown(context.system).addTask(CoordinatedShutdown.PhaseServiceRequestsDone, "complete hdfs sinks") { () =>
    validQueue.complete()
    invalidQueue.complete()
  }

您还可以使用共享终止开关 (https://doc.akka.io/docs/akka/current/stream/stream-dynamic.html#sharedkillswitch)，您可以使用 .via(sharedKillSwitch.flow) 将其放入您的流程中，您可以从 CoordinatedShutdown 关闭开关：

// create it somewhere, use in your flows
val sharedKillSwitch = KillSwitches.shared("hdfs-switch")

// use switch in CoordinatedShutdown
CoordinatedShutdown(context.system).addTask(CoordinatedShutdown.PhaseServiceRequestsDone, "complete hdfs sinks") { () =>
    sharedKillSwitch.shutdown()
  }