为什么这个 Akka Streams 应用程序异常退出？答案

【问题标题】：Why does this Akka Streams application exit abnormally?为什么这个 Akka Streams 应用程序异常退出？
【发布时间】：2018-08-07 05:22:15
【问题描述】：

我正在通过 Packt 的 Akka Cookbook 模块了解 Akka Streams。我运行TransformingStreamsApplication.scala 示例并得到这个：

为了让actor系统在流处理完成后退出，我添加了如下回调：

// Future[IOResult]
val future = stream.run()
future.onComplete(_ => system.terminate())

但是这次应用直接退出，没有任何控制台输出：

我想出的解决方法是添加Thread.sleep(10000)：

我想解释一下这种行为。

【问题讨论】：

标签： scala akka future akka-stream

【解决方案1】：

这是你referencing的信息流：

val stream = FileIO.fromPath(path)
  ...
  .to(Sink.foreach(println))

由于使用了to 连接器，对上述流调用run() 会返回源的具体化值，在本例中为Future[IOResult]。发生的情况是您在流元素到达接收器之前终止了 Actor 系统。

不要添加Thread.sleep，而是使用toMat 和Keep.right 更改流以产生接收器的具体化值。这个物化值也是一个Future，一旦这个Future完成，你就可以终止actor系统：

val stream = FileIO.fromPath(path)
  ...
  .toMat(Sink.foreach(println))(Keep.right)

val future = stream.run()
future.onComplete(_ => system.terminate())

请注意，有一个速记方法，称为runWith：

val stream = FileIO.fromPath(path)
  ...
  .runWith(Sink.foreach(println))

stream.onComplete(_ => system.terminate())

【讨论】：