Scala Future concurrency問題

Question

以下是我的課程，我同時運行任務。我的問題是，即使獲得所有功能的結果，我的應用程序也永遠不會結束。我懷疑線程池並沒有關閉，即使在執行任務之後，我的應用程序仍然處於活動狀態。相信我，我已經使用了很多功能，但沒有成功。我在這裏錯過了什麼？

import scala.concurrent.ExecutionContext.Implicits.global 
    import scala.concurrent.Future 
    import scala.collection.mutable.ListBuffer 
    import scala.util.Failure 
    import scala.util.Success 

    object AppLauncher{ 

     def launchAll(): ListBuffer[Future[String]] = { 
     // My code logic where I launch all my threads say 50 
     null 
     } 

def main(args:Array[String]):Unit= { 
register(launchAll()) 
} 



     def register(futureList: ListBuffer[Future[String]]): Unit = 
     { 
      futureList.foreach { future => 
      { 
       future.onComplete { 
       case Success(successResult) => { 
        println(successResult) 
       } 
       case Failure(failureResult) => { println(failureResult) } 
       } 
      } 
      } 
     } 
    } 

Answer 1

最後我終於明白了這個問題。這個問題是有道理的，因爲即使在我的期貨成功完成後，線程池也沒有被終止。我試圖通過稍微改變我的實現來區分這個問題，如下所示。

添加突出顯示的代碼，以關閉池之前//進口scala.concurrent.ExecutionContext.Implicits.global

import scala.concurrent.Future 
    import scala.collection.mutable.ListBuffer 
    import scala.util.Failure 
    import scala.util.Success 

    //Added ExecutionContex explicitly 
    import java.util.concurrent.Executors 
    import concurrent.ExecutionContext 

    object AppLauncher { 

    //Implemented EC explicitly 
    private val pool = Executors.newFixedThreadPool(1000) 
    private implicit val executionContext = ExecutionContext.fromExecutorService(pool) 

    def launchAll(): ListBuffer[Future[String]] = { 
     // My code logic where I launch all my threads say 50 
     null 
    } 

    def main(args: Array[String]): Unit = { 
     register(launchAll()) 
    } 

    def register(futureList: ListBuffer[Future[String]]): Unit = 
     { 
     futureList.foreach { future => 
      { 

      println("Waiting...") 
      val result = Await.result(future, scala.concurrent.duration.Duration.Inf) 

      println(result) 

      } 

     } 

 pool.shutdownNow() 
     executionContext.shutdownNow() 

 println(pool.isTerminated() + " Pool terminated") 
     println(pool.isShutdown() + " Pool shutdown") 

     println(executionContext.isTerminated() + " executionContext terminated") 
     println(executionContext.isShutdown() + " executionContext shutdown") 
     } 

    } 

結果

假游泳池終止

真正的泳池關閉

假的ExecutionContext終止

真正的ExecutionContext關機

將突出顯示的代碼解決了我的問題後。我確保了我的代碼中沒有資源泄漏。我的場景允許我在所有未來完成時殺死池。我意識到我將優雅的回調實現改爲阻止實現，但它仍然解決了我的問題。

Answer 2

通常情況下，當你的Future秒的迭代工作，你應該使用Future.sequence從而改變比方說，一個Seq[Future[T]]到Future[Seq[T]]。

因此，使用這樣的：

def register(futureList: Seq[Future[String]]) = Future.sequence(futureList) foreach { results => 
    println("received result") 
} 

，如果你想爲它完成，你也可以做的東西線到每個未來和打印輸出映射;

def register(futureList: Seq[Future[String]]) = Future.sequence (
    futureList.map(f => f.map { v => 
    println(s"$v is complete") 
    v 
    })) map { vs => 
    println("all values done $vs") 
    vs 
}