Scala Future

Future

• scala.concurrent.Future 異步執行代碼塊

  import java.time._
  import scala.concurrent._
  import ExecutionContext.Implicits.global // 全局線程池
  Future {
          Thread.sleep(10000)
          println(s"This is the future at ${LocalTime.now}")
  }
  println(s"This is the present at ${LocalTime.now}")

• 監聽結果（阻塞）

  import scala.concurrent.duration._
  val f = Future { Thread.sleep(10000); 42 }
  val result = Await.result(f, 10.seconds) //阻塞10s
  
  val f = Future { ... }
  Await.ready(f, 10.seconds)
  val Some(t): Option[Try[T]] = f.value
  
  t match {
          case Success(v) => println(s"The answer is $v")
          case Failure(ex) => println(ex.getMessage)
  }

ready()

• 到達等待時間無結果時，會拋出異常 TimeoutException
• 任務拋出的異常時，result() 會再次拋出異常， ready() 可獲取結果

• 回調

  val f = Future { 
          Thread.sleep(10000)
          if (random() < 0.5) throw new Exception
          42
  }
  f.onComplete {
          case Success(v) => println(s"The answer is $v")
          case Failure(ex) => println(ex.getMessage)
  }

• 問題：1.回調地獄；2.執行順序沒法預知

  val future1 = Future { getData1() }
  val future2 = Future { getData2() }
  future1 onComplete {
          case Success(n1) =>
                  future2 onComplete {
                          case Success(n2) => {
                                  val n = n1 + n2
                                          println(s"Result: $n")
                                  }
                          case Failure(ex) => ...
                  }
          case Failure(ex) => ...
  }

  將 Future 看做集合

  // val 會當即執行，def 調用時執行
  val future1 = Future { getData1() }
  val future2 = Future { getData2() }
  // 都獲取到結果時，纔會進行計算
  val combined = for (n1 <- future1; n2 <- future2) yield n1 + n2

• Promise
  • 與 Java 8 中的 CompletableFuture 相似

  • Future 只讀，在任務完成時隱式設置結果值；Promise 相似，但結果值可顯式設置

    // Future
    def computeAnswer(arg: String) = Future {
        val n = workHard(arg)
        n
    }
    
    // Promise
    def computeAnswer(arg: String) = {
        val p = Promise[Int]()
        Future {
            val n = workHard(arg)
            // 顯式設置結果
            p.success(n)
            workOnSomethingElse()
        }
        // 當即返回該 Promise 對應的 Future
        p.future
    }
    
    // 多個任務對應一個 Promise
    val p = Promise[Int]()
    Future {
        var n = workHard(arg)
        // 若 Promise 未完成則接受結果並返回 true；不然忽略結果並返回 false
        p.trySuccess(n)
    }
    Future {
        var n = workSmart(arg)
        p.trySuccess(n)
    }

• 執行上下文
  • 默認執行在全局的 fork-join 線程池（默認大小爲核數），適用於計算密集型任務

  • 對於阻塞型/IO密集型的任務，可以使用 Java 的 Executors

    // 隱式聲明，或者使用 Future.apply 顯式聲明
    val pool = Executors.newCachedThreadPool()
    implicit val ec = ExecutionContext.fromExecutor(pool)
    
    val f = Future {
        val url = ...
        blocking {
            val contents = Source.fromURL(url).mkString
            ...
        }
    }