Future　scalaで非同期演算を扱いたい時に使う

未完了、失敗、成功の状態がある

基本

Future(Thread.sleep(1000 * 10)) // Future(<not completed>) 未完了
Future(1) // Future(Success(1))　成功 
Future(1/0) // Future(Failure(java.lang.ArithmeticException: / by zero)) 失敗、例外を保持
Future(throw new OutOfMemoryError) 
//java.lang.OutOfMemoryError 致命的な例外　ExecutionContextのreportFailureが呼ばれその後
//Thread.uncaughtexceptionhandlerが呼ばれる.　Futureは完了しない

※ Fatal（致命的）な例外とは以下の例外のもの

VirtualMachineError,
- OutOfMemoryとか
ThreadDeath,
- スレッドが死んだらどうしようもない
InterruptedException,
- 割り込み要求されたら中断して投げるやつ
LinkageError,
- AbstractMethodErrorなど
ControlThrowable,
- Breakable.breakとか
NotImplementedError


trait Future[+T] extends Awaitable[T]

Future traitに定義されているメソッドはAwaitableにあるreadyとresultメソッド以外は全部ノンブロッキング

Futureのメソッドたち

成功, 失敗したら副作用を実行

成功時

Future(1).foreach(println) // １

失敗時

Future(1/0).failed.foreach(println) // java.lang.ArithmeticException: / by zero

＊　onSuccess, onFailureはScala 2.12から非奨励

成功も失敗も両方扱いたい(戻り値はUnit)

Future(1 / 0).andThen {
  case Success(x) => println(x)
  case Failure(_: ArithmeticException) => println("/ by zero")
}

計算結果を使って処理したい

Future(1).map(_ + 1) // Future(Success(2))

途中で失敗したら後続の処理はskipしてそのまま結果が帰る

Future(1).map(_ / 0).map(_ + 1) // Future(Failure(java.lang.ArithmeticException: / by zero))

計算結果をFutureを返す関数に渡して合成したい

def futureFunc(i: Int): Future[Int] = ???
Future(1).flatMap(futureFunc)

Futureを返す関数が複数合成したい時はfor式を使える

for {
 x1 <- futureFunc1()
 x2 <- futureFunc2(x1)
 x3 <- futureFunc3(x2)
} yield x3

ただしfutureFunc2はfutureFunc1が完了したあとに実行 futureFunc3はfutureFunc2が完了した後に実行される事に注意(並列に実行される訳ではない)

Futureを並列に実行して合成したい

合成したいものが二つの場合

(Future(1) zip Future(2)).map { case (x, y) => x + y } // Future(Success(3))

scala 2.12では

(Future(1) zipWith Future(2))((x, y) => x + y) // Future(Success(3))

と書けるようになった

二つより多い時
scalazの|@|を使うと便利

import scalaz._
(Future(1) |@| Future(2) |@| Future(3))(_ + _ + _) // Future(Success(6))

catsではこう書く

import cats._
import cats.implicits._
(Future(1), Future(2), Future(3)).mapN(_ + _ + _) // Future(Success(6))

個数が決まってないものはFuture.sequenceを使う

//Seq[Future[A]]を Future[Seq[A]]に変換 
Future.sequence(Seq(Future(1), Future(2), Future(3)))// Future(Success(List(1, 2, 3))

特定の例外を回復したい時

recover

Future(1 / 0).recover { 
  case _: ArithmeticException => 0  
}.map(_ + 1) //Future(Success(1))

成功も失敗も両方扱いたい(戻り値あり)

Future(1).transform(_ + 1, ex => ex)

このtransform　成功は成功　失敗は失敗にしか変換できず使いづらかったが
2.12以降は　

transform[T](f: Try[U] => Try[T])

とTryを受け取りTryを返せるようになったので

Future(1).transform { 
 case Success(i) => Success(i + 1)
 case Failure(e) => Success(0)
}

失敗　成功両方を扱えみたいに書けるようになり実用的になった

Futureのコンパニオンオブジェクトに生えている便利メソッド達

Future.successful

//Futureを実行したくないが型は合わせたい時
futureFunc1().flatMap { i => 
  if (i > 0) 
    futureFunc2(i)
  else 
    Future.successful(0)
}

Future.sequence

//Seq[Future[A]]を Future[Seq[A]]に変換 
Future.sequence(Seq(Future(1), Future(2), Future(3)))// Future(Success(List(1, 2, 3))

//失敗したものあった時には失敗が帰る 
Future.sequence(Seq(Future(1), Future(throw new Exception), Future(3)))// Future(Failure(java.lang.Exception))

Future.firstCompletedOf

//最初に完了したものを返す
Future.firstCompletedOf(Seq(Future { Thread.sleep(1000); 1 }, Future(2))) //Future(Success(2))

//最初に完了したものが失敗だったら失敗を返す
Future.firstCompletedOf(Seq(Future { Thread.sleep(1000); 1 }, Future(throw new Exception()))) //Future(Failure(java.lang.Exception))

Future.find

//条件に合うものを返す
Future.find(List(Future { Thread.sleep(1000); 2 }, Future(4)))(_ % 2 == 0) //Future(Success(Some(2)))

//失敗したものは無視する
Future.find(List(Future { Thread.sleep(1000); 2 }, Future(throw new Exception)))(_ % 2 == 0) //Future(Success(Some(2)))

scala 2.12で　これまでのfind(以前は失敗を無視しないで報告していていた)は非奨励になった

Future.traverse

//Seq(1, 2).map(Future(_)).sequnceと同じ
Future.traverse(Seq(1, 2))(Future(_)) // Future(Success(List(1, 2, 3))

scala 2.12から
fold, reduceが非奨励になり代わりにfoldLeft, reduceLeftが増えた。

Future.unit

Future.successfull(())

を

Future.unit

と書けるようになった

また永遠に計算を完了しない

Future.never

もできた

ExecutionContext

Futureの各メソッドでimplicit parameterとして要求されるもの
Futureは何を計算するかのWhatを表すがExecutionContextはどのように並列実行するかのHowを表す
基本スレッドプールのようなものでデフォルトはjavaのFork/Joinフレームワーク
- import scala.concurrent.ExecutionContext.Implicits.global
- akka actorのデフォルトもForkJoinPool

ExecutionContextの定義はシンプル
executeに渡されるRunnableがFutureに渡されていた計算
def execute(runnable: Runnable)　= runnable.runとかすると同期Futureとか作れる
＊prepareは無視していいです

trait ExecutionContext {

  def execute(runnable: Runnable): Unit


  def reportFailure(@deprecatedName('t) cause: Throwable): Unit


  def prepare(): ExecutionContext = this

}

代表的なスレッドプール

固定スレッドプールの概念図

キューに溜まったtaskを待機スレッドに渡し実行する　わかり易いモデル
しかし問題点も。。

並列数（＝スレッド）を増やすとtaskキューへのロックが多発して性能が劣化する
スレッドを増やしても並列数はcpuのcore数分
スレッドを増やしすぎるとコンテキストスイッチが多発して性能劣化(=C10K問題)

つまり　従来の１リクエスト＝１スレッドでは同時接続数に上限がある

そこでfork-join-pool

スレッド数がコア数×N個ほどなのでコンテキストスイッチが少ない
各スレッドがキューを持っているのでロック獲得待ちが少ない
暇なスレッドが出来たら他のスレッドからjobを奪う（work-stealing）のするのでバランスよくjobが振り分けられる
１リクエスト＝１スレッドではないのでスレッド数の上限が同時接続数の上限ではなくなる！

無敵に見えるが向き不向きがある。。

問題

import scala.concurrent.ExecutionContext.Implicits.global
(1 to 8).map(_ => Future {Thread.sleep(1000 * 30)})
Future(1)

論理コアが8のcpuで最後のFuture(1)の計算は最低で何秒かかるでしょうか

答え　最低でも30秒

ForkJoinPoolのデフォルトのスレッド数が論理コア数なのでスレッドを論理コア数分ブロックしてしまうと全体が止まる
ForkJoinが得意なものはスレッドをブロックしない処理、重い計算やブロッキングIOは基本的にやるべきではない (例えばakka-actorだと全actorが止まってメッセージを処理出来なくってしまう)
ここでいうブロッキング処理とはブロッキングIO、高価な計算、 Thread.sleep, synchronizedなどによるスレッドの資源獲得待ちのこと

といっても全てのIOがノンブロッキングではないし全ての計算をノンブロッキングな粒度まで分解できる訳ではない
ではForkJoinPoolでブロッキングする処理を扱いたい時にはどうするか

方法1
ForkJoinPool限定だがblockingメソッドでスレッドをブロックする処理を囲めばForkJoinPoolが必要と判断した場合に
Threadを新規に作りブロックしないようにしてくれる

import scala.concurrent._
(1 to 8).map(_ => Future(blocking {Thread.sleep(1000 * 30)}))

が新規にスレッドを作るのはコストが高いし必ずしもスレッドを作ってくれる訳ではない

方法２
王道　
固定スレッドプールのExecutionContextを作ってそこでブロックする処理を書く

val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(num))
(1 to 8).map(_ => Future {Thread.sleep(1000 * 30)}(ec))

これでForkJoinを使用しているFutureはブロックされないようになる
しかし結局　固定スレッドプールをコンテキストスイッチが多発して性能は劣化する
（そもそも論としてマシンあたりの同時接続数をそこまで要求しない普通のwebアプリの場合固定スレッドプールの方が扱いやすいのかもしれない）

まとめ

FutureはmapやflatMapを使って合成できる
Future objectには便利なメソッドが生えている
デフォルトのfork-joinスレッドプールをブロックしてはいけない
ブロッキングをしたい時は使用するスレッドプールを吟味する
Futureは用法を守って正しく服用する

ScalaのFuture