未来是不完整的

Question

object Executor extends App {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  implicit val ec = system.dispatcher
  import akka.stream.io._
  val file = new File("res/AdviceAnimals.tsv")
  import akka.stream.io.Implicits._
  val foreach: Future[Long] = SynchronousFileSource(file)
    .to( Sink.outputStream(()=>System.out))
    .run()
  foreach onComplete { v =>
    println(s"the foreach is ${v.get}")  // the will not be print
  }
}

但是如果我将Sink.outputStream(()=>System.out)改为Sink.ignore, println(s"the foreach is ${v.get}")将打印。

谁能解释一下为什么?

Answer 1

您没有等待流完成，相反，您的主方法(Executor的主体)将完成，并且由于主方法完成退出JVM被关闭。

你要做的，是阻塞该线程，并且在未来完成之前不退出应用程序。

object Executor extends App {
  // ...your stuff with streams...
  val yourFuture: Future[Long] = ???
  val result = Await.result(yourFuture, 5 seconds)
  println(s"the foreach is ${result}")
  // stop the actor system (or it will keep the app alive)
  system.terminate()
}

Answer 2

巧合的是，我创建了几乎相同的应用程序来测试/播放Akka Streams。是输入的隐式诱发了这个问题吗?这个应用程序对我来说很好:

object PrintAllInFile extends App {
  val file = new java.io.File("data.txt")
  implicit val system = ActorSystem("test")
  implicit val mat    = ActorMaterializer()
  implicit val ec     = system.dispatcher
  SynchronousFileSource(file)
    .to(Sink.outputStream(() => System.out))
    .run()
    .onComplete(_ => system.shutdown())
}

注意在'onComplete'中停止ActorSystem。