Scala将来の並行処理の問題

Question

以下は、私がタスクを同時に実行するクラスです。私の問題は、すべての機能の結果が得られてもアプリケーションが終了しないということです。私はスレッドプールが私の仕事の後でも私のアプリケーションを生きているリードしている停止されていないと思われる。私はそれを把握するためにたくさんのグーグルだが運がない。ここで私は何が欠けているのですか？あなたがFuture秒の反復可能で動作するとき

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

    } 

結果シャットダウン・プールに

0123を強調し、コードを追加する前に

偽プールが

真プールのシャットダウンに

を終了し、偽のExecutionContextがハイライトされたコードを追加した後

真のExecutionContextのシャットダウンに

を終了し、私の問題を解決しました。私のコードでリソースリークがないことを保証しました。私のシナリオでは、すべての先物が完了したときにプールを殺すことができます。私は、エレガントなコールバック実装をブロッキング実装に変更したが、それでも私の問題を解決したという事実を認識しています。

Answer 2

は通常、あなたはFuture[Seq[T]]に言う変わるFuture.sequence、Seq[Future[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 
}