The Future is not complete?

问题:

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
  }
}

but if I change the Sink.outputStream(()=>System.out) to Sink.ignore, the println(s"the foreach is ${v.get}") will print.

Can somebody explain why?

回答1:

You are not waiting for the stream to complete, instead, your main method (the body of Executor) will complete, and since the main method is done exits the JVM is shut down.

What you want to do, is to block that thread and not exit the app before the future completes.

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()
}

回答2:

Coincidently I created almost the same app for testing/playing with Akka Streams. Could the imported implicits cause the problem? This app works fine for me:

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())
}

Note the stopping of the ActorSystem in the 'onComplete'. Otherwise the app will not exit.