Futureで非同期に処理を実行する方法

Scala 3.3.1

最終更新：2023年12月7日

この記事では、Future を使用して、非同期的（並列）に処理を実行する方法について解説します。

非同期処理をするには Future を使う

非同期的に処理を行いたいとき、つまり既存の処理に並列して処理を実行したいときには、Future を使用します。

非同期的に処理を行いたい場面としては、例えばディスクIO、ネットワークIO、あるいはコストの大きい計算など、比較的時間のかかる処理を行う場合です。

もっとも簡単な使用例は以下のようになります。

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

Future:
  for (i <- 1 to 3)
    Thread.sleep(100)
    println(s"future: $i") // A

println("executed") // B

このコードでは、Future のあとに続くブロック内に時間のかかる処理を記述しています。

実行結果は以下のようになります。

executed
future: 1
future: 2
future: 3

コードの書かれている順番としては(B)の println("executed") がいちばん最後ですが、文字列 "executed" は最初に出力されています。

ここでは、Futureブロック内の処理とブロック外の処理が並列に処理されることとなった結果、ブロック外のprintln (B)が先に実行され、ブロック内のprintln (A)は遅れて実行されています。

ScalaにおいてはこのようにしてFutureを使って非同期的（並列）に処理を行うことができます。

ただし、以下の説明を読まずにこのコードをそのまま使ってプロダクションコードを書いた場合には、予期せぬスピードダウンが発生するかもしれません。

ぜひこの記事の最後まで読んでくださいね。

使い方の基本は `Future` と `ExecutionContext` の組み合わせ

Scalaで非同期処理をするには、 Future と、 ExecutionContext （実行コンテキスト）について知る必要があります。

「実行コンテキスト」とは何でしょうか？一言でいうとスレッドプールのことです（後述）。

まずは通常の処理がどのように行われているかについて振り返ってみましょう。

通常の処理では、標準の「実行コンテキスト」を使用している

上述のコードから、Futureブロックを剥がして同期的に実行してみます。

for (i <- 1 to 3)
  Thread.sleep(100)
  println(s"future: $i") // A
println("executed") // B

これらの処理は上から順次、つまり同期的に処理されます。

したがって実行結果は以下のようになります。

future: 1
future: 2
future: 3
executed

(A)がすべて実行されたのち(B)が出力されています。

実際のところ、特に指定しない限り、これらすべての処理はこのコードを実行した「実行コンテキスト」内の「スレッド」において実行されています。

実行コンテキストとは、処理をどのように実行するか定義したグループのことです。

実行の仕組みについては後述しますが、内部に「スレッド」という名の処理レーンが1本以上用意されており、このレーン上で処理が実行されることになっています。

このスレッドのグループのことを「スレッドプール」と呼びます。 ExecutionContextとはスレッドプールのことだと考えてよいです。

通常の処理はこの「標準の実行コンテキスト」内で実行されるので、明示的に実行コンテキストを指定する必要はありません。

実行コンテキストの存在を意識せずに処理を実行できているのは、このためです。

非同期処理では、何らかの実行コンテキストを用意する必要がある

そもそも、実行コンテキストは「コンテキスト（脈絡）」という名の通り、複数存在し得ます。つまり、標準以外の実行コンテキストを作成して使用することもできます。

通常のように順次処理している分には、実行コンテキストは自明でした。これ対して非同期処理においては、どのような実行コンテキストにおいて処理を行うべきかが自明ではありません。

なぜ複数種類の実行コンテキストを使用したいことがあるのでしょうか。それは、高効率なプログラムを書く場合には、扱う処理に対して最適な実装の実行コンテキストを用意して使い分ける必要があるためです。

したがって、使用者が何らかの実行コンテキストを用意し、Futureに対してわかるように指定してあげる必要があるというわけです。

さて、上述の、Futureを使って非同期的に処理をしているコードに戻ってみましょう。

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

Future:
  for (i <- 1 to 3)
    Thread.sleep(100)
    println(s"future: $i") // A

println("executed") // B

このコードでは、冒頭の2行で何かをインポートしていますね。 scala.concurrent.Future と scala.concurrent.ExecutionContext.Implicits.global です。 2つ目こそ、Scalaが標準で用意している「実行コンテキスト」です。

つまり、このコードでは、Scalaが標準で用意している実行コンテキストを使用して非同期処理を行うよう指定しているというわけです。

Scalaが用意している ExecutionContext には scala.concurrent.ExecutionContext.global と scala.concurrent.ExecutionContext.Implicits.global があります。これらのglobalはほぼ同じものを指しています。 ExecutionContext.global は、標準の実行コンテキストそのものです。 ExecutionContext.Implicits.global はそのエイリアスで、違いはimplicitであることだけです。インポートすると暗黙変数として使用することができます。

Futureを使用している箇所においてExecutionContext.Implicits.globalをインポートすることで、 Futureに対して標準の実行コンテキストにて処理を実行するよう暗黙的に指定したこととなります。

使い方は「`Future`に手を突っ込んで中身を操作する」イメージ

一度Futureで包んだら、中身の値を操作したい場合にもできるだけFutureの外から触るようにして、できるだけ最後まで中身を取り出さないでおくようにします。

なぜこのような扱いをするのかというと、Futureの値を取り出すには非同期処理の完了を待つ必要があるためです。

取り出した後は同期的に処理されることになるため、一連の流れの途中で取り出してしまうと処理効率が低下しがちです。したがって、Futureから値を取り出すのはできるだけ控えましょう。

難しそうに聞こえますが、Futureには便利なメソッドがたくさん用意されているため、値を取り出さないまま外から様々な操作を加えることができます。

`Future`の中身は`map`や`flatMap`メソッドで操作する

mapやflatMapメソッドを使用すると、Futureの中身の値を操作することができます。

まずはmapメソッドを使用して中身を操作してみましょう。

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

val f: Future[Int] = Future:
  Thread.sleep(100)
  100

f.map(_ * 3) // mapメソッドを使用する
  .foreach(println)

このコードでは、もとの値100をmapメソッドによって3倍しています。

出力結果は以下のようになります。

あるいは、Futureの中にFutureが入れ子になっている場合、以下のようにflattenメソッド使うとうまく扱うことができます。

val f: Future[Future[Int]] = Future:
  Thread.sleep(100)
  Future:
    Thread.sleep(100)
    100

val flatF: Future[Int] = f.flatten // flattenメソッドを使用する

flatF.map(_ * 3)
  .foreach(println)

出力結果は以下のようになります（mapのときと同様です）。

また、そもそも入れ子になってしまうと扱いづらいです。入れ子になるのを防ぎたい場合には、以下のようにflatMapメソッド使って処理することができます。

flatMapメソッドは、flattenメソッドとmapメソッドを組み合わせたものです。

val f: Future[Int] = Future {
  Thread.sleep(100)
}.flatMap { _ =>
    Future:
      Thread.sleep(100)
      100
}

f.map(_ * 3) // flatMapメソッドを使用する
  .foreach(println)

出力結果は以下のようになります（map、flattenのときと同様です）。

このようにして、入れ子になって積み上がりそうなFutureを適度に潰して平たくして処理したい場合には flatMap メソッドを使用すると便利です。

複数の`Future`を扱うならfor式を使う

複数のFutureを扱いたい場合には、for式を使うと便利です。

val f1: Future[Int] = Future(100)
val f2: Future[Int] = Future(3)
for {
  a <- f1
  b <- f2
} yield
  println(a * b)

出力結果は以下のようになります

for式は、内部的にはflatMap、map、あるいはwithFilterメソッドに変換されて実行されます。

便利メソッドを活用する

Futureには、mapやflatMap以外にも多くの便利メソッドが用意されています。

上で軽く振れたwithFilterメソッドもそうです。

APIドキュメントを参照して、どのようなメソッドがあるのかを確認しておきましょう。

外部リンク：Futureトレイト

外部リンク：Futureオブジェクト

3つの注意点を押さえる

Futureを使用する際に注意すべき点が大きく3つあります。

実行順
実行コンテキスト
例外処理

最後に、この3つについて確認してみましょう。

実行順を把握する

実はFutureは、非同期的（並列）に実行されるとはいえ、一旦変数に置かないと並列処理されません。

以下の例を見てください。この例では、Futureを返すメソッドを2回呼んでいます。

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

def getF(num: Int): Future[Unit] = Future:
  for (i <- 1 to 5)
    Thread.sleep(100)
    println(s"future $num: $i")

for {
  _ <- getF(1)
  _ <- getF(2)
} yield
  println("finished")

Futureを使っているので、getF(1)とgetF(2)が並列に実行されてほしいと思う人が多いでしょう。

しかしresult1の結果は以下のようになります。

future 1: 1
future 1: 2
future 1: 3
future 1: 4
future 1: 5
future 2: 1
future 2: 2
future 2: 3
future 2: 4
future 2: 5
finished

どうも様子がおかしいですね。仕組みとしては、getF(1)の結果を待ってからgetF(2)が実行されているので、このようなことが起こります。たとえワイルドカード_を使って結果の値を使わないようにしていても、この現象が起こるので注意してください。

さて、もう一つ例を用意しました。こちらは、メソッド呼び出しの結果のFutureをそれぞれ一旦変数に置いてからfor式に渡しています。

val f1: Future[Unit] = getF(1)
val f2: Future[Unit] = getF(2)
for {
  _ <- f1
  _ <- f2
} yield
  println("finished")

一旦変数に束縛しているのが冗長に見えますが、挙動の違いを確認してみましょう。

result2の結果は例えば以下のようになります。

future 1: 1
future 2: 1
future 1: 2
future 2: 2
future 1: 3
future 2: 3
future 1: 4
future 2: 4
future 1: 5
future 2: 5
finished

こんどは順不同に、つまり並列に実行されていることがわかります（実際には、f1の出力とf2の出力が必ず交互に並ぶとは限りません）。変数に束縛した時点でそれぞれFutureブロック内の処理が開始されているので、並列に処理できていたというわけです。

広いスコープにおいてはFutureを変数に置かないことはあまりないので心配する必要性は薄いのですが、狭いスコープ内で処理を確実に並列に実行したい場合には注意してください。

また、for式そのものは並列に処理するための仕組みではないことにも注意してください。

最適な実行コンテキスト（`ExecutionContext`）を選ぶ

非同期処理を行うということは効率よく処理を行いたいということです。

効率よく処理を行うには、実行コンテキストに関する理解は欠かせません。

内部実装について知る

実行コンテキストは、内部実装的には「スレッドプール」です。

標準の実行コンテキスト（`ExecutionContext.global`）は使わないようにする

Scalaに標準で用意されている実行コンテキストExecutionContext.globalは、デフォルトでは並列数がCPUのコア数分だけになっています。

例えばコアが1つだけの場合、Futureで時間のかかる処理を実行すると、一発でスレッドプールが詰まります。並列もなにもありません。

あるいは時間のかかる処理がコア数以上に同時に実行された場合にも詰まります。

単純に ExecutionContext.globalを使って非同期処理をしてもいい場面は、並列処理数がコア数を超えない場合のみです。

sbt consoleなどで手元でちょっと実験するためのものと割り切ることをおすすめします。

Scala ExecutionContextって何 / Futureはスレッド立ち上げじゃないよ - ましめも https://mashi.hatenablog.com/entry/2014/11/24/010417

Future内でThread.sleepはするな - ましめも https://mashi.hatenablog.com/entry/2014/12/08/000149

フレームワークがExecutionContextを用意してくれる場合もある

ExecutionContextは、フレームワークが用意してくれることもあります。

Akkaや、AkkaHTTPを使用しているPlayFrameworkなどでは、Akka自体がActorSystemを用意してくれています。

ActorSystemはExecutionContextを継承しています。

Akkaではapplication.confにて詳しく設定することができます。

ExecutionContextを自前で用意する

手頃な実行コンテキストが用意されていない場合は、自分で用意しましょう。

用意する方法は大きく2つあります。

JavaのAPIでスレッドプールを作成し、ExecutionContext#fromExecutorや#fromExecutorServiceメソッドに渡してExecutionContextを得る
システムプロパティを設定し、ExecutionContext.globalの並列数をCPUのn倍にする
- ExecutionContext.global自体が事故の元なので、おすすめはしません

タスクの種類によってExecutionContextを適切に使い分ける必要があります。

ExecutionContextの種類について確認する

Javaで作成できるスレッドプールには以下のようなものがあります。

ForkJoinPool (Java SE 15 & JDK 15) https://docs.oracle.com/javase/jp/15/docs/api/java.base/java/util/concurrent/ForkJoinPool.html

※ scala.concurrent.ExecutionContext.globalは、実装上は「CPUコア数×n倍」を並列数としたForkJoinPoolです。

また、java.util.concurrentExecutorsの以下のようなメソッドを使用して作成することもできます（抜粋）。

newCachedThreadPool()
newFixedThreadPool(int nThreads)
newScheduledThreadPool(int corePoolSize)
newSingleThreadExecutor()
newSingleThreadScheduledExecutor()
newWorkStealingPool()

Executors (Java SE 15 & JDK 15) https://docs.oracle.com/javase/jp/15/docs/api/java.base/java/util/concurrent/Executors.html

外部リンク：scala.concurrent.ExecutionContext.global

外部リンク：scala.concurrent.ExecutionContext.Implicits.global

外部リンク：Futures and Promises | Scala Documentation

外部リンク：scala.concurrent.ExecutionContext

必ずエラーハンドリングをする

エラー処理に関しては、何もしなければ結果を取り出すことがないというFutureの特徴が災いする可能性がありますので、注意して扱いましょう。

そもそも同期処理の場合には、例外発生時には呼び出し元が処理をします。

catchするなどして対応しなければ処理全体が失敗する原因となりますが、逆に言えば対応を半ば強いられるという点で、異常発生に気づきやすい仕組みになっています。

これに対して、Futureブロック内で例外が発生した場合には、一旦Future型が例外をキャッチします。

このままでは実はキャッチされた例外は誰にも処理されず、最終的にガベージコレクションによって宇宙の彼方へと消え去ることになります。

「例外が握りつぶされれていることに長い間運用していても気づかない」などということが起こり得るので、忘れずにエラーハンドリングをしましょう。

`Future`での例外処理には `onComplete` や `recover` を使う

Future内部で生じた例外を処理するには、onCompleteメソッド、またはrecover/recoverWithメソッドを使用します。

onCompleteメソッドは戻り値がUnitです。したがって、エラーハンドリングが最終結果に影響しない場合に使用します。

val f = Future:
  throw new Exception("error!")

f.onComplete:
  case Success(_) => println(1)
  case Failure(_) => println(100)

出力結果は以下のようになります。

これに対して、recover/recoverWithメソッドは戻り値がFuture[U]です。例外処理をしてもとの処理に復帰させたいときに使用します。

val f = Future {
  throw new Exception("error!")
}.recover:
  case _ => 100

f.foreach(println)

recoverメソッドとrecoverWithメソッドの使い分けは、引数としてFutureに包まれた値をとるかどうかです。

val f = Future {
  throw new Exception("error!")
}.recoverWith:
  case _ => Future(100)

f.foreach(println)

Akkaでは、例外処理していない場合にはログを出力してくれる

ちなみに、Akkaにおいては、ExecutionContextとしてAkkaのActorSystemを使用している限り、その実装であるActorSystemImplが未処理の例外をログに出力してくれるので、ログさえ見ておけば例外発生が握りつぶされてしまうことはありません。

[AD] Scalapediaに広告を出してみませんか？

サイト内検索

カテゴリ「並行並列処理」の記事

Futureで非同期処理をする方法 Futureで非同期に処理を実行する方法

カテゴリ一覧

はじめに

Scala 3（Dotty）特集

Scala 3で廃止予定の機能

ニュース・その他のトピック環境構築

JDK

sbt

IDE

IntelliJ IDEA

エディタ

基礎知識・基本文法中級文法文字列処理コレクション

List

Map

Play Framework入門 Spring Boot入門ビルドツールデータベース並行並列処理その他のユーティリティシステム Java開発経験者のためのScala入門 Javaについてライブラリその他

カテゴリ「並行並列処理」の記事

Futureで非同期処理をする方法 Futureで非同期に処理を実行する方法

Futureで非同期に処理を実行する方法

非同期処理をするには Future を使う

使い方の基本は Future と ExecutionContext の組み合わせ

通常の処理では、標準の「実行コンテキスト」を使用している

非同期処理では、何らかの実行コンテキストを用意する必要がある

使い方は「Futureに手を突っ込んで中身を操作する」イメージ

Futureの中身はmapやflatMapメソッドで操作する

複数のFutureを扱うならfor式を使う

便利メソッドを活用する

3つの注意点を押さえる

実行順を把握する

最適な実行コンテキスト（ExecutionContext）を選ぶ

内部実装について知る

標準の実行コンテキスト（ExecutionContext.global）は使わないようにする

フレームワークがExecutionContextを用意してくれる場合もある

ExecutionContextを自前で用意する

ExecutionContextの種類について確認する

必ずエラーハンドリングをする

Futureでの例外処理には onComplete や recover を使う

Akkaでは、例外処理していない場合にはログを出力してくれる

サイト内検索

カテゴリ「並行並列処理」の記事

カテゴリ一覧

カテゴリ「並行並列処理」の記事

使い方の基本は `Future` と `ExecutionContext` の組み合わせ

使い方は「`Future`に手を突っ込んで中身を操作する」イメージ

`Future`の中身は`map`や`flatMap`メソッドで操作する

複数の`Future`を扱うならfor式を使う

最適な実行コンテキスト（`ExecutionContext`）を選ぶ

標準の実行コンテキスト（`ExecutionContext.global`）は使わないようにする

`Future`での例外処理には `onComplete` や `recover` を使う