実戦中のPromiseとFuture

前章では、Futureタイプと、高可読性、高組合せ性の非同期実行コードを記述する方法について説明します.
Futureは謎の一部にすぎません.読み取り専用タイプで、計算した値を使用したり、計算中に発生したエラーを処理したりすることができます.しかし、その前に、この値を入れる方法が必要です.この章では、Promiseタイプでこの目的を達成する方法について説明します.
タイプPromise
以前は、scala.concurrentのfutureメソッドに順次実行されるコードブロックを渡し、機能ドメインにExecutionContextが与えられ、コードブロックを不思議に非同期で呼び出し、Futureタイプの結果を返す.
このFutureを取得する方法は簡単ですが、Futureインスタンスを作成して埋め込む方法は他にもあります.これがPromiseです.PromiseではFutureに値を入れることができますが、一度しかできません.Futureが完了すると、変更できません.
1つのFutureインスタンスは、常に1つのPromiseインスタンスに関連付けられています.REPLでfutureメソッドを呼び出すと、戻ってくるのもPromiseであることがわかります.

import concurrent.Future
import concurrent.Future

scala> import concurrent.future
import concurrent.future

scala> import concurrent.ExecutionContext.Implicits.global
import concurrent.ExecutionContext.Implicits.global

scala> val f: Future[String] = future { "Hello World!" }
f: scala.concurrent.Future[String] = scala.concurrent.impl.Promise$DefaultPromise@2b509249

あなたが得た対象はDefaultPromiseで、それはFutureとPromiseインタフェースを実現しましたが、これは具体的な実現の詳細です(訳注、興味のある読者はその実現のソースコードをめくることができます)、使用者はコード実現がFutureと対応するPromiseの間のつながりをはっきりさせることを知っているだけです.
この小さな例は,Promise以外にFutureを完成させる方法はなく,future方法も補助関数にすぎず,具体的な実装メカニズムを隠していることを示している.
では、Promiseタイプを直接使用する方法を見てみましょう.
承諾を与える
約束が果たされるかどうかについて話したとき、よく知られている例は政治家の選挙約束だった.
推選された政治家が投票者に減税の約束をしたと仮定する.これはPromise[TaxCut]で表すことができます.

import concurrent.Promise
case class TaxCut(reduction: Int)
// either give the type as a type parameter to the factory method:
val taxcut = Promise[TaxCut]()
// or give the compiler a hint by specifying the type of your val:
val taxcut2: Promise[TaxCut] = Promise()
// taxcut: scala.concurrent.Promise[TaxCut] = scala.concurrent.impl.Promise$DefaultPromise@66ae2a84
// taxcut2: scala.concurrent.Promise[TaxCut] = scala.concurrent.impl.Promise$DefaultPromise@346974c6

このPromiseを作成すると、futureメソッドを呼び出して、承諾の将来を得ることができます.

 val taxCutF: Future[TaxCut] = taxcut.future
 // `> scala.concurrent.Future[TaxCut] `  scala.concurrent.impl.Promise$DefaultPromise@66ae2a84

返されるFutureはPromiseと同じではないかもしれませんが、同じPromiseでfutureメソッドを呼び出すと、PromiseとFutureの間の1対1の関係を確保するために、常に同じオブジェクトが返されます.
終了承諾
いったん約束をして、世界が遠くない将来にそれを実行すると言ったら、できるだけ実現したほうがいいです.Scalaでは、成功しても失敗してもPromiseを終了できます.
約束を果たす.
Promiseを正常に終了するには、successメソッドを呼び出し、期待された結果を伝えることができます.

  taxcut.success(TaxCut(20))

そうすると、Promiseは他の値を書き込むことができなくなり、あえて書き直すと異常が発生します.
このときPromiseに関連付けられたFutureも成功裏に完了し、登録されたコールバックが実行されたり、このFutureがマッピングされたりしているので、そのときにはマッピング関数も実行されるはずです.
一般に,Promiseの完了と返されるFutureに対する処理は異なるスレッドで発生する.Promiseを作成し、すぐに呼び出し元に関連付けられたFutureを返す可能性がありますが、実際には別のスレッドが計算しています.
この点を説明するために、減税を例に挙げます.

object Government {
  def redeemCampaignPledge(): Future[TaxCut] = {
    val p = Promise[TaxCut]()
    Future {
      println("Starting the new legislative period.")
      Thread.sleep(2000)
      p.success(TaxCut(20))
      println("We reduced the taxes! You must reelect us!!!!1111")
    }
    p.future
  }
}

この例ではFuture伴生オブジェクトを使用していますが、混同されないようにしましょう.この例のポイントは、Promiseが呼び出し者のスレッドで完成したわけではありません.
今、私たちは当初の選挙宣言を実行し、FutureにonCompleteのコールバックを追加します.

import scala.util.{Success, Failure}
val taxCutF: Future[TaxCut] = Government.redeemCampaignPledge()
println("Now that they're elected, let's see if they remember their promises...")
taxCutF.onComplete {
  case Success(TaxCut(reduction)) =>
    println(s"A miracle! They really cut our taxes by $reduction percentage points!")
  case Failure(ex) =>
    println(s"They broke their promises! Again! Because of a ${ex.getMessage}")
}

この例を複数回実行すると、ディスプレイ出力の結果順序が不確定であり、最終コールバック関数が実行され、成功したcaseに入ることがわかります.
約束にそむく
政治家は約束に背くことに慣れており、Scalaプログラマーもそうするしかないことがある.failureメソッドを呼び出し、例外を渡し、Promiseを終了します.

case class LameExcuse(msg: String) extends Exception(msg)
object Government {
  def redeemCampaignPledge(): Future[TaxCut] = {
     val p = Promise[TaxCut]()
     Future {
       println("Starting the new legislative period.")
       Thread.sleep(2000)
       p.failure(LameExcuse("global economy crisis"))
       println("We didn't fulfill our promises, but surely they'll understand.")
     }
     p.future
   }
}

このredeemCampaignPledge実現は最終的に約束に反する.このPromiseをfailureで終了すると、successメソッドのように再書き込みできません.関連するFutureもFailureで終わる.
Tryが既に存在する場合は、Promiseのcompleteメソッドに直接渡すことで、これを終了することができます.このTryがSuccessであれば、関連するFutureは正常に完了し、そうでなければ失敗します.
Futureベースのプログラミング実践
Futureベースのプログラミングパターンを使用してアプリケーションの拡張性を向上させるには、アプリケーションを下から上まで非ブロックモードに設計する必要があります.これは、基本的にアプリケーション層のすべての関数が非同期であり、Futureを返すべきであることを意味する.
現在,可能な使用シーンの一つはWebアプリケーションの開発である.人気のあるScala Webフレームワークでは、応答が完了してから戻るのではなく、応答をFuture[Response]として返すことができます.これは、Webサーバがより多くの接続を少量のスレッドで処理できるようにするため、非常に重要です.サーバFuture[Response]の機能を付与することで、サーバスレッドプールの利用率を最大化できます.
また、アプリケーションのサービスは、データベース・レイヤや(または)一部の外部サービスを複数回呼び出す必要がある場合があります.この場合、複数のFutureを取得し、for文で新しいFutureに組み合わせることができ、簡単に読めます.最終的には、Webレイヤは、このようなFutureをFuture[Response]に変更する.
しかし、どのように実践の中でこれらを実現すればいいのでしょうか.3つの異なるシーンを考慮する必要があります.
非ブロックIO
アプリケーションは多くのIO操作に関連する可能性が高い.たとえば、データベースと対話する必要がある場合や、クライアントとして他のWebサービスを呼び出すこともできます.
もしそうであれば、Java非ブロックIOに基づいて実装されたいくつかのライブラリを使用してもよいし、直接またはNettyのようなライブラリを介してJavaのNIO APIを使用してもよい.このようなライブラリは、大量の接続を定量的スレッドプールで処理できます.
しかし、このようなライブラリを開発したい場合は、Promiseと直接付き合うのが適切です.
ブロックIO
NIOベースのライブラリがない場合があります.たとえば、Java世界のほとんどのデータベースドライバは、ブロックIOを使用しています.Webアプリケーションでは、このようなドライバでデータベースへの大量アクセスの呼び出しを開始すると、これらの呼び出しはサーバスレッドで発生していることを覚えておいてください.この問題を回避するために、データベースと対話する必要があるすべてのコードをfutureコードブロックに入れることができます.

// get back a Future[ResultSet] or something similar:
Future {
  queryDB(query)
}

これまで,暗黙的に利用可能なグローバルExecutionContextを用いてこれらのコードブロックを実行してきた.通常、より良い方法は、データベース・レイヤに専用のExecutionContextを作成することです.JavaのExecutorServiceから使用できます.これは、アプリケーションの他の部分が影響を受けないように、スレッドプールを非同期で調整してデータベース呼び出しを実行できることを意味します.

import java.util.concurrent.Executors
import concurrent.ExecutionContext
val executorService = Executors.newFixedThreadPool(4)
val executionContext = ExecutionContext.fromExecutorService(executorService)

長時間運転の計算
アプリケーションの本質的な特徴に応じて、1つのアプリケーションは、IO(CPUが密集しているタスク)に全く関与しない長時間実行されるタスクを呼び出すこともあります.これらのタスクは、サーバスレッドで実行する必要はありません.そのため、Futureにする必要があります.

Future {
  longRunningComputation(data, moreData)
}

同様に、これらのCPUが密集している計算を処理するために、いくつかの専用ExecutionContextが望ましい.これらのスレッドプールのサイズをどのように調整するかは、適用の特徴に依存し、これらは本明細書の範囲を超えている.