12
12

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@KenyaSaitoh(斉藤 賢哉)

3.2 Executorフレームワークと非同期タスク～新Java Advanced編 / AI対応版

12
Last updated at Posted at 2023-02-04

はじめに

自己紹介

皆さん、こんにちは、Udemy講師の斉藤賢哉です。私はこれまで、約30年にわたって企業システムの開発に携わってきました。現在も現役のアーキテクトとして、重要システムの技術設計やソリューションの選定を担当しています。直近では、システム開発へのAI適用（AI駆動開発）という全社的な施策でリーダーを務めています。
セミナーでの登壇や雑誌への技術記事の寄稿を行ってきたほか、以下の書籍も執筆しています。

Udemy講座のご紹介

この記事の内容は、私が講師を務めるUdemy講座『新 Java Advanced編 / AI対応版』の一部の範囲をカバーしたものです。『新 Java Advanced編 / AI対応版』はこちらのリンクから購入できます。

この講座は、以下のような皆様にお薦めします。

  • Javaの基礎を学び終え、設計とレビューができるエンジニアを目指す方
  • AIが書いた並行処理や入出力のコードに潜むリスクを、仕組みから見抜けるようになりたい方
  • フレームワークの裏で使われるリフレクションやアノテーションの仕組みを理解したい方
  • 「Oracle認定Javaプログラマ」Goldの範囲を、仕組みから理解したい方

AIとチャットできるオールインワンの学習環境をリリースしました

Udemy講座の演習を動かすための学習環境として、Windows向けのデスクトップアプリ「Codinable（コーディナブル）」をリリースしました。
エディタ、実行結果、テスト結果、SQL、Webブラウザ、そしてAIチャットを1つの画面にまとめた、オールインワンの学習環境です。

Codinableの画面構成

学習に必要なものは、すべてCodinableの中で完結します。

  • 演習の実行環境（Java 25、Gradle、Bashなど）が一式同梱されており、JDKのインストールや環境変数の設定は不要です
  • インストールして「新規コース開始・再開」からコースを選ぶだけで、講座のすべての演習がそのまま動きます
  • 書き換えて動かなくなっても、「初期化」で配布時の状態に戻せます
  • 開いている演習のファイルと実行結果を踏まえて、AIに質問したり（Askモード）、ファイルの書き換えを指示したり（Agentモード）できます

AIに書き換えを指示し、差分を確かめてから実行する、というAI駆動開発の基本の流れを、Codinableの中だけで安全に体験できます。

AIチャットに書き換えを指示し、差分を確かめて実行する

AIチャットは、受講者ご自身が生成AIサービス（Anthropic、OpenAI、Googleのいずれか）のAPIキーを登録して使うBYOK（Bring Your Own Key）方式です。利用料金は各社から直接請求されます。APIキーを登録しなくても、AIチャット以外の機能はすべて使えます。

Codinableのインストーラは、以下のGitHubのReleasesページからダウンロードできます（64ビット版のWindows 10／11、管理者権限は不要）。

この記事を含むシリーズ全体像

この記事はJava SEの一部の機能・仕様を取り上げたものですが、一連のシリーズになっており、シリーズ全体でJava SEを網羅しています。また認定資格である「Oracle認定Javaプログラマ」（Silver、Gold）の範囲もカバーしています。シリーズの全体像および「Oracle認定Javaプログラマ」の範囲との対応関係については、以下を参照ください。

3.2 Executorフレームワークと非同期タスク

チャプターの概要

チャプター3.1では、スレッドとメモリーの関係、競合状態や同期化といった、マルチスレッドの土台となる仕組みを取り上げました。本チャプターでは、スレッドを直接生成せずにタスクの実行を任せるExecutorフレームワークを取り上げます。スレッドプールの種類と選び方、ExecutorServiceのシャットダウン、ScheduledExecutorServiceによる定期実行、FutureとCallableによる結果と例外の受け取り、複数タスクの一括実行を、タイムラインの図で整理します。AIが書いたコードでは、プールの選択、シャットダウンの漏れ、例外の握りつぶし、待機のタイムアウトの欠落が典型的な問題になります。これらを見抜けるようになることが、本チャプターの目標です。

3.2.1 Executorフレームワークの構造

本レッスンでは、Executorフレームワークが「タスク」と「それを実行するスレッド」をどのように分離しているかを整理します。

スレッドプールによるタスクの実行

並行処理ユーティリティ（java.util.concurrentパッケージ）は、マルチスレッドを効率的かつ安全に実現するためのクラス群で、その中心がExecutorフレームワークです。アプリケーションはRunnableやCallableでタスクを作って投入するだけで、どのスレッドで実行するかはExecutorに任せます。

タスクごとにスレッドを生成する場合と、スレッドプールを使う場合を比べます。

タスクごとのスレッド生成とスレッドプールの比較

  • 左の「タスクごとにnew Thread()」では、タスク1〜5のそれぞれにスレッドが生成され、終わると破棄される
  • 左の下に「生成のコストが毎回かかる・同時に動くスレッドの数に上限が無い」と注記がある
  • 右の「スレッドプール」では、タスク1〜5が待ち行列に並び、あらかじめ用意された3つのスレッドが順に取り出して実行する
  • 右の下に「スレッドを使い回す・同時に動く数をプールの大きさで制限できる」と注記がある

スレッドプールにタスクが割り当てられる様子を、もう少し詳しく示します。

スレッドプールによるタスクの割り当てと使い回し

  • 実行するタスクは、待ち行列に並んで順番を待つ
  • スレッドプールには、あらかじめスレッド①、スレッド②、スレッド③が用意されている
  • 先頭のタスク1は、空いているスレッド①に割り当てられて実行される
  • タスクを終えたスレッドはプールに戻り、次のタスクに使い回される

主要なインタフェースとクラス

Executorフレームワークを構成する主要なインタフェースとクラスの関係を、図で示します。

Executorフレームワークを構成する主要なインタフェースとクラス

  • ExecutorServiceは、Executorを継承したインタフェースである
  • ScheduledExecutorServiceは、ExecutorServiceを継承したインタフェースである
  • ThreadPoolExecutorクラスはExecutorServiceを、ScheduledThreadPoolExecutorクラスはScheduledExecutorServiceを実装する
  • ScheduledThreadPoolExecutorは、ThreadPoolExecutorを継承したクラスである
  • Executorsクラスは、ExecutorServiceとScheduledExecutorServiceのインスタンスを生成して返す

コードで扱うのは主にExecutorServiceとScheduledExecutorServiceの2つのインタフェースで、インスタンスはExecutorsのスタティックメソッドから取得します。

【表】ExecutorServiceインタフェースの主要なAPI

API（メソッド） 説明
submit(Runnable)／submit(Callable<T>) タスクを投入し、すぐにFutureを返す（レッスン3.2.5参照）
invokeAll(Collection<? extends Callable<T>>) 複数のタスクを一括で実行し、すべて完了するとFutureのリストを返す
shutdown() 新規タスクの受け付けを止める。投入済みのタスクは続ける
shutdownNow() 実行中のタスクに割り込みを送り、未開始のタスクのリストを返す
awaitTermination(long, TimeUnit) すべてのタスクの終了を指定時間だけ待つ。時間内に終わらなければfalse
close() shutdown()の後、すべてのタスクの終了を待つ（Java 19以降。try-with-resources文で使える）

TimeUnitは時間の単位を表す列挙型で、MILLISECONDS、SECONDS、MINUTESなどの値を持ちます。

3.2.2 スレッドプールの種類と選び方

本レッスンでは、Executorsが返すスレッドプールの違いを、同じタスクを実行したときのタイムラインで比べます。AIが選んだプールが、タスクの性質と資源の制約に合っているかを判断できるようになります。

プールの種類とタイムライン

共通のタスクとして、RunnableTaskを使います。指定したカウントの回数だけ1秒ずつ停止しながら値を3倍し、カウントが4なら約4秒かかるタスクです。

【表】ExecutorServiceを取得するためのExecutorsクラスのAPI

API（メソッド） スレッドの数 待ち行列 向いている用途
newSingleThreadExecutor() 1つ（使い回す） 上限なし 順序を守って1つずつ実行したい処理
newFixedThreadPool(int) 指定した数で固定 上限なし CPUやデータベース接続など、資源に合わせて並列度を抑えたい処理
newCachedThreadPool() 必要に応じて増やす（上限なし）。60秒使われないと減る 持たない 短いタスクが散発的に来る処理
newVirtualThreadPerTaskExecutor() タスクごとに新しい仮想スレッド 持たない 待ち時間の長いI/Oを大量に同時実行する処理（Java 21以降）

単一のスレッドのプールに、4秒かかるタスクを3つ投入するコードです。

  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_2.Main_Single
ExecutorService executor = Executors.newSingleThreadExecutor(); //【1】
RunnableTask fooTask = new RunnableTask("foo", 4);
RunnableTask barTask = new RunnableTask("bar", 4);
RunnableTask bazTask = new RunnableTask("baz", 4);
executor.submit(fooTask); //【2】
executor.submit(barTask);
executor.submit(bazTask);
System.out.println("[ Main ] finish");
  • 【1】スレッドを1つだけ持つExecutorServiceを取得します。
  • 【2】submit()はタスクの終了を待たずに戻るため、[ Main ] finishが先に表示されます。
  • このコードはshutdown()を呼んでいないため、3つのタスクが終わってもプログラムは終了しません（レッスン3.2.3参照）。

単一のスレッドでは、3つのタスクが順に実行されます。

newSingleThreadExecutor()による3つのタスクの実行

  • 横軸は時間（0秒〜12秒）で、上からメインスレッド、スレッド1の帯が並ぶ
  • 0秒の時点で、メインスレッドがfooTask、barTask、bazTask（各4秒）を投入し、すぐに「[ Main ] finish」を表示する
  • スレッド1では、fooTask（0〜4秒）、barTask（4〜8秒）、bazTask（8〜12秒）が順に実行される
  • 3つのタスクのスレッドIDは同じで、全体で約12秒かかる

Main_Cachedでは同じ3つのタスク（8秒、15秒、5秒）を動的なプールで実行し、3つのスレッドが生成されて約15秒で終わります。Main_Fixedではスレッドを2つに固定したプールで実行します。

  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_2.Main_Fixed
ExecutorService executor = Executors.newFixedThreadPool(2);
RunnableTask fooTask = new RunnableTask("foo", 8);
RunnableTask barTask = new RunnableTask("bar", 15);
RunnableTask bazTask = new RunnableTask("baz", 5);
// タスクの投入は前の例と同じため省略

固定的なスレッドプールで3つのタスクが実行される時間の流れを、図で示します。

固定的なスレッドプール（スレッド2つ）による3つのタスクの実行

  • メインスレッドは、0秒の時点でfooTask（8秒）、barTask（15秒）、bazTask（5秒）の3つのタスクを投入する
  • スレッド1でfooTask、スレッド2でbarTaskが並列に実行される
  • bazTaskは、空いているスレッドがないため待機する
  • 8秒の時点でfooTaskが終わると、空いたスレッド1にbazTaskが割り当てられる
  • barTaskが終わる15秒の時点で、すべてのタスクが終わる

上限の無いプールと待ち行列の危険

表のとおり、newFixedThreadPool()の待ち行列と、newCachedThreadPool()のスレッドの数には上限がありません。投入の速さが処理の速さを上回り続けると、どちらもメモリーを使い果たします。

2つのプールで、何が際限なく増えるのかを図で比べます。

固定プールと動的プールで際限なく増えるもの

  • 左の「newFixedThreadPool(2)」では、スレッドは2つのままで、処理しきれないタスクが待ち行列に際限なく積み上がる（ラベル「待ち行列が伸び続ける→メモリー不足」）
  • 右の「newCachedThreadPool()」では、待ち行列は無く、タスクが来るたびにスレッドが増え続ける（ラベル「スレッドが増え続ける→メモリー・CPUの枯渇」）
  • 上に、どちらにも「毎秒100件のタスクが投入され、毎秒50件しか処理できない」という同じ条件が書かれている
  • 下に「上限が必要ならThreadPoolExecutorのコンストラクタで、スレッド数と容量の決まった待ち行列を指定する」と注記がある

仮想スレッドによるタスク実行

newVirtualThreadPerTaskExecutor()は、投入されたタスクごとに新しい仮想スレッドを割り当てます（仮想スレッドの仕組みはレッスン3.1.7参照）。

try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(new RunnableTask("foo", 8));
    executor.submit(new RunnableTask("bar", 15));
    executor.submit(new RunnableTask("baz", 5));
} // ブロックを抜けるときにclose()が呼ばれ、すべてのタスクの終了を待つ
  • 3つのタスクに、それぞれ新しい仮想スレッドが割り当てられ、並列に実行されます。
  • try-with-resources文を抜けるときにclose()が呼ばれ、すべてのタスクの終了を待ちます。

レビューの観点

  • プールの種類が、タスクの性質（CPUを使い続けるか、I/Oで待つか）と資源の制約（データベースの接続数など）に合っているか。CPU中心ならスレッド数をコア数程度に固定する
  • リクエストごと・メソッドの呼び出しごとにExecutors.newXxx()でプールを作っていないか。プールはアプリケーションで共有して使い回す
  • 際限なくタスクが投入され得る場面で、上限の無い待ち行列やスレッド数のまま使っていないか
  • 投入したタスクがチャプター3.1の意味で共有データに触れていないか。Executorを使ってもスレッドセーフになるわけではない

3.2.3 ExecutorServiceのシャットダウン

本レッスンでは、ExecutorServiceが終了するまでの状態と、shutdown()・shutdownNow()・awaitTermination()の役割を整理します。終わらないプログラムや、止まらないタスクの原因を見抜けるようになります。

シャットダウンの状態遷移

ExecutorServiceは、タスクを一度受け付けると、その後も次のタスクを待ち続けます。終了させるにはshutdown()かshutdownNow()を呼びます。

ExecutorServiceがたどる状態を、図で整理します。

ExecutorServiceのシャットダウンの状態遷移

  • 「稼働中」の状態では、新しいタスクを受け付け、投入済みのタスクを実行する
  • shutdown()を呼ぶと「シャットダウン中」になり、新しいタスクは拒否されるが、待ち行列のタスクも含めて最後まで実行される
  • shutdownNow()を呼ぶと、待ち行列の未開始のタスクが取り除かれて呼び出し元に返され、実行中のタスクに割り込みが送られる
  • すべてのタスクが終わると「終了」になり、awaitTermination()で待っていたスレッドにtrueが返る
  • 下に「割り込みに応じないタスクは、shutdownNow()の後も実行を続ける」と注記がある

タイムアウト付きの終了待ち

shutdown()の後、一定時間待っても終わらなければshutdownNow()で割り込む、という形が定石です。

  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_2.Main_Await
// fooTask（8秒）、barTask（15秒）、bazTask（5秒）の投入は省略
executor.shutdown(); //【1】
try {
    if (! executor.awaitTermination(12L, TimeUnit.SECONDS)) { //【2】
        executor.shutdownNow(); //【3】
    }
} catch(InterruptedException ie) {
    executor.shutdownNow();
}
System.out.println("[ Main ] finish");
  • 【1】新規タスクの受け付けを閉じます。投入済みの3つのタスクは処理を続けます。
  • 【2】すべてのタスクの終了を、最大12秒待ちます。時間内に終わればtrueが返ります。
  • 【3】12秒を超えるとfalseが返り、shutdownNow()で実行中のタスクに割り込みを送ります。

このコードの時間の流れを、タイムラインで確かめます。

awaitTermination()とshutdownNow()のタイムライン

  • 横軸は時間（0秒〜15秒）で、上からメインスレッド、fooTask（8秒）、barTask（15秒）、bazTask（5秒）の帯が並ぶ
  • 0秒でメインスレッドがshutdown()を呼び、awaitTermination(12秒)で待機に入る
  • bazTaskは5秒、fooTaskは8秒の時点で正常に終わる
  • 12秒の時点でタイムアウトし、メインスレッドがshutdownNow()を呼んで、barTaskに割り込みが届く
  • barTaskはThread.sleep()の待機中にInterruptedExceptionを受け、15秒を待たずに終了する

shutdownNow()は割り込みを送るだけで、タスクを強制的に止めるものではありません。割り込みに応答しないタスク（レッスン3.1.2参照）は、そのまま実行を続けます。

レビューの観点

  • ExecutorServiceを生成したコードに、shutdown()（またはtry-with-resources文によるclose()）が対になっているか。無いとプログラムが終了しない、またはスレッドが漏れる
  • shutdown()の直後に結果を使っていないか。shutdown()は終了を待たない。待つならawaitTermination()やclose()を使う
  • awaitTermination()の戻り値を無視していないか。falseのとき（終わっていないとき）の扱いが決まっているか
  • shutdownNow()で止めたいタスクが、割り込みに応答する作りになっているか

3.2.4 ScheduledExecutorServiceによる定期実行

本レッスンでは、遅延実行と定期実行を行うScheduledExecutorServiceを取り上げます。「周期に合わせる」か「間隔を空ける」かの違いと、定期実行のタスクが例外で静かに止まる危険を判断できるようになります。

周期に合わせた実行と間隔を空けた実行

ScheduledExecutorServiceは、Executors.newScheduledThreadPool(int)などで取得します。

【表】ScheduledExecutorServiceインタフェースの主要なAPI

API（メソッド） 説明
schedule(タスク, 遅延, 単位) 指定時間だけ遅延して、一回だけ実行する（RunnableとCallableの両方を受け付ける）
scheduleAtFixedRate(タスク, 初回の遅延, 周期, 単位) 開始の時刻が周期に合うように繰り返す
scheduleWithFixedDelay(タスク, 初回の遅延, 間隔, 単位) 前回の終了から指定の間隔を空けて繰り返す
  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_3.Main_FixedRate
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
RunnableTask2 task = new RunnableTask2();
// 2秒遅れて開始し、10秒毎に繰り返しタスクを実行する
executor.scheduleAtFixedRate(task, 2L, 10L, TimeUnit.SECONDS);
  • RunnableTask2は、呼び出されるたびに7秒、14秒、4秒、10秒と、異なる時間がかかるように作られたタスクです。
  • 前回のタスクが周期を超えた場合、次のタスクは待たずに直ちに実行されます。同じタスクが重なって並列に実行されることはありません。

scheduleAtFixedRate()による実行のタイミングを、図で示します。

scheduleAtFixedRate()による周期に合わせた繰り返し実行

  • scheduleAtFixedRate()には、タスク、開始までの遅延時間（2秒）、周期（10秒）、時間の単位を渡す
  • 1回目のタスクは2秒の遅延後に開始し、7秒で終わる
  • 2回目のタスクは、次の予定時刻（1回目の開始から10秒）まで3秒待機してから開始する
  • 2回目のタスクは14秒かかって次の予定時刻（1回目の開始から20秒）を過ぎるため、待機せずに直ちに3回目を実行する（並列実行はされない）
  • 3回目のタスクは4秒で終わるため、次の予定時刻（1回目の開始から30秒）まで2秒待機してから4回目を開始する

lsn_3_2_3.Main_FixedDelayは、scheduleWithFixedDelay(task, 2L, 5L, TimeUnit.SECONDS)で、終了から5秒の間隔を空けて繰り返します。同じタスクを2つの方法で実行した場合を、タイムラインで比べます。

scheduleAtFixedRate()とscheduleWithFixedDelay()の比較

  • 上段のscheduleAtFixedRate（周期10秒）では、予定時刻が2秒、12秒、22秒、32秒…と周期にそろい、タスクの長さによって待ち時間が変わる（2回目が予定時刻の22秒を過ぎたため、3回目は2回目の終了の26秒に直ちに始まる）
  • 下段のscheduleWithFixedDelay（間隔5秒）では、7秒のタスクの終了から5秒、14秒のタスクの終了から5秒と、終了と次の開始の間がいつも5秒になる
  • 下段では、タスクの処理時間が変わると、開始の時刻がその分ずれていく
  • 下に「定時に動かしたい→scheduleAtFixedRate()」「処理の後に必ず休みを入れたい→scheduleWithFixedDelay()」と使い分けがある

例外で止まる定期実行

定期実行のタスクで例外が送出されると、それ以降の実行は行われなくなります。例外はどこにも表示されず、返されたScheduledFutureのget()を呼んだときに初めて分かります。

定期実行のタスクが例外で止まる流れ

  • 1回目と2回目のタスクは正常に終わる
  • 3回目のタスクでRuntimeExceptionが送出される
  • 4回目以降の予定時刻が来ても、タスクは実行されない（点線の枠で「実行されない」）
  • ログには何も出ず、プログラムは動き続けているように見える
  • 下に「良い例: タスクの本体をtryで囲み、例外をログに出してから続ける」と注記がある

レビューの観点

  • 定期実行のタスクの本体がtry-catchで囲まれ、例外がログに出るようになっているか。囲まれていないと、1回の例外で以後の実行が静かに止まる
  • 「周期に合わせる」と「間隔を空ける」のどちらが要件に合うかを確かめる。処理が周期より長くなり得る場合の動きも確かめる
  • shutdown()されないまま、アプリケーションの終了を妨げていないか

3.2.5 FutureとCallableによる結果と例外の受け取り

本レッスンでは、非同期に実行したタスクの結果と例外を、Futureを介して受け取る仕組みを取り上げます。待ち方（ブロッキング、タイムアウト）と例外の受け取り方を、レビューで確かめられるようになります。

Callableとsubmit()の戻り値

Callable<V>は、Runnableと同じく非同期に実行するタスクを表し、V call() throws Exceptionで結果を返したり、チェック例外を送出したりできます。submit()はタスクの終了を待たず、すぐにFutureを返します。Futureは「後で結果が入る引換券」のようなものです。

【表】Futureインタフェースの主要なAPI

API（メソッド） 説明
V get() タスクの完了まで待ち、結果を返す。タスクの例外はExecutionExceptionに包まれて送出される
V get(long, TimeUnit) 指定時間を超えるとTimeoutExceptionを送出する
boolean isDone() タスクが完了していればtrueを返す
boolean cancel(boolean) タスクを取り消す。引数がtrueなら実行中のタスクに割り込みを送る
  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_4.callable.Main_Get_3
Future<Integer> future = executor.submit(task); //【1】8秒かかる処理を起動
doSomething(5); //【2】5秒間、何らかの処理を行う
Integer result = future.get(); //【3】待機してから結果を取得する（3秒待機）
  • 【1】taskはCallable<Integer>を実装したCallableTaskで、8秒かけて3の8乗の6561を返します。
  • 【2】結果を受け取る前に、メインスレッドで別の処理を進めます。
  • 【3】get()はタスクの完了まで待ちます。全体の所要時間は13秒ではなく約8秒です。

メインスレッドとタスクの時間の流れを、図で示します。

submit()からget()までの間にメインスレッドで別の処理を行う流れ

  • メインスレッドのMain_Get_3は、executor.submit()でCallableTaskを起動する
  • CallableTaskは、タスクを8秒かけて実行する
  • その間、メインスレッドは何らかの別の処理を5秒間行う
  • メインスレッドはfuture.get()を呼び出し、タスクの実行終了まで3秒待機する
  • 8秒の時点で、メインスレッドが実行結果を受け取る

例外の受け取り

call()で送出された例外は、タスクを実行したスレッドではなくFutureの中に保存され、get()を呼んだときにExecutionExceptionに包まれて送出されます。元の例外はgetCause()で取り出します（例はlsn_3_2_4.exception.Main）。

例外がタスクから呼び出し元へ届く経路を、図で確かめます。

タスクの例外がFutureを経由して届く経路

  • 右のプールのスレッドで、call()がIllegalStateExceptionを送出する
  • 例外はスレッドの外へは伝わらず、Futureの中に「失敗」として保存される（スレッドはプールに戻る）
  • 左のメインスレッドがfuture.get()を呼ぶと、ExecutionException（原因: IllegalStateException）が送出される
  • 下に「get()を一度も呼ばないと、例外はどこにも表示されずに消える」と注記がある

Runnableをsubmit()した場合も同じで、run()の例外はFutureの中に保存されます。戻り値を捨てると、タスクの失敗に誰も気付けません。

// 問題のあるコード：Futureを捨てているため、例外が消える
executor.submit(() -> importData());

// 良いコード：結果を確かめ、例外の原因を扱う
Future<?> f = executor.submit(() -> importData());
try {
    f.get(30, TimeUnit.SECONDS);
} catch (ExecutionException e) {
    log.error("取り込みに失敗", e.getCause());
} // InterruptedException、TimeoutExceptionの扱いは省略
  • importData()とlogは説明のための仮のメソッドとロガーです。
  • 結果を使わない場合でも、失敗を検出したいならget()で確かめるか、タスクの中で例外をログに出します。

完了待機のタイムアウト

get()は、タスクが終わらなければ永久に待ちます。get(long, TimeUnit)で上限を設けると、時間を超えた時点でTimeoutExceptionが送出されます。

  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_4.callable.Main_Timeout
Future<Integer> future = executor.submit(task); //【1】8秒かかる処理を起動
Integer result = future.get(4, TimeUnit.SECONDS); //【2】4秒でタイムアウト
  • 【1】8秒かかるタスクを起動します。
  • 【2】4秒でTimeoutExceptionが送出されます。get()の3種類のチェック例外は、catchで扱います（省略）。

タイムアウトの後、タスクがどうなるかを図で示します。

get()のタイムアウトとタスクの継続

  • 横軸は時間（0秒〜8秒）で、上がメインスレッド、下がCallableTaskの帯である
  • メインスレッドは0秒からget(4秒)で待ち、4秒の時点でTimeoutExceptionを受けて先へ進む
  • CallableTaskは4秒を過ぎても止まらず、8秒まで実行を続ける（ラベル「結果は誰にも受け取られない」）
  • 下の別案では、4秒の時点でfuture.cancel(true)を呼び、CallableTaskに割り込みが届いて終了する

isDone()を使うと、待たずに完了を確かめながら別の処理を進められます（例はlsn_3_2_4.callable.Main_Done）。

レビューの観点

  • submit()の戻り値のFutureを捨てていないか。捨てる場合、タスクの中で例外をログに出しているか
  • get()にタイムアウトが付いているか。外部のシステムを呼ぶタスクのget()が無期限になっていないか
  • タイムアウトした後、不要になったタスクをcancel(true)で止めているか
  • ExecutionExceptionをcatchしたとき、getCause()で元の例外を扱っているか。包んだ例外のメッセージだけをログに出していないか
  • submit()の直後にget()を呼んでいないか。それなら非同期にする意味が無い

3.2.6 複数タスクの一括実行と分割処理

本レッスンでは、invokeAll()による複数タスクの一括実行と、データを分割して並列に集計する方法を取り上げます。

invokeAll()による一括実行

invokeAll()は、Callableのコレクションを受け取って一括で実行し、すべて完了してからFutureのリストを返します。

  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_5.invokeall.Main
List<CallableTask> taskList = new ArrayList<CallableTask>();
taskList.add(new CallableTask("foo", 8)); //【1】
taskList.add(new CallableTask("bar", 15));
taskList.add(new CallableTask("baz", 5));
List<Future<Integer>> futureList = executor.invokeAll(taskList); //【2】
for (Future<Integer> future : futureList) { //【3】
    System.out.println("result => " + future.get());
}
  • 【1】3つのタスクをリストに格納します。
  • 【2】newCachedThreadPool()のプールで3つのタスクが並列に実行され、すべて終わるまで待ちます。
  • 【3】結果は投入したリストと同じ順序（6561、14348907、243）で並びます。すべて完了しているため、get()で待つことはありません。

一括実行の時間の流れを、タイムラインで確かめます。

invokeAll()による一括実行のタイムライン

  • 横軸は時間（0秒〜15秒）で、上からメインスレッド、fooTask（8秒）、barTask（15秒）、bazTask（5秒）の帯が並ぶ
  • 0秒でメインスレッドがinvokeAll()を呼び、3つのタスクが同時に始まる
  • メインスレッドの帯は、0秒から15秒まで「invokeAll()で待機」と表示される
  • bazTaskは5秒、fooTaskは8秒で終わるが、メインスレッドは最も長いbarTaskの終わる15秒まで戻らない
  • 15秒の時点で、メインスレッドが投入順の結果のリストを受け取る

コレクションの分割処理

巨大なコレクションを集計するときは、分割してタスクに分担させ、最後に結果を合計することで、スループットを高められます。ここでは1から10万までのリストを10個に分割し、10個のSumTaskで並列に集計します。

コレクションを分割して並列に集計する流れを、図で示します。

コレクションの分割処理による並列集計

  • 1〜100000のコレクションを、10個のコレクションに分割する

  • 分割されたコレクション①〜⑩を、非同期タスク①〜⑩がそれぞれ並列に集計する

  • 各非同期タスクの集計値を、さらに集計する

  • 10個の集計値の合計が、コレクション全体の集計値になる

  • 03_thread_concurrent: pro.kensait.java.advanced.lsn_3_2_5.collection.Main

int paraCount = 10;
List<List<Integer>> splitedListList =
        splitIntegerList(intList, paraCount); //【1】
List<Callable<Integer>> taskList = new ArrayList<>();
for (List<Integer> splitedList : splitedListList) {
    taskList.add(new SumTask(splitedList)); //【2】
}
ExecutorService executor = Executors.newFixedThreadPool(paraCount);
List<Future<Integer>> futureList = executor.invokeAll(taskList); //【3】
long sum = 0L;
for (Future<Integer> future : futureList) {
    sum += future.get(); //【4】
}
  • 【1】1から10万までのintListを、ユーティリティメソッドで10個に分割します。
  • 【2】分割したリストごとに、合計をIntegerで返すSumTask（Callable<Integer>）を作ります。
  • 【3】10個のタスクを並列に実行し、すべての完了を待ちます。
  • 【4】各タスクの結果を合計し、5000050000を得ます。この値はintの範囲を超えるため、sumはlongにしています。

各タスクは自分に割り当てられた部分リストだけを読み、結果は戻り値で返しています。タスクの間で書き換える共有データが無いため、同期化は不要です。この「分割して、共有せずに計算し、最後に合わせる」形は、並列ストリーム（チャプター4.2参照）でも使われています。

レビューの観点

  • 並列のタスクが、共有のリストやカウンターに結果を書き込んでいないか。結果は戻り値で返し、最後に合わせる形になっているか
  • 集計の途中や結果の型が、桁あふれしないか（intで足りるか）
  • 分割した部分リストが、元のリストのビュー（subList()など）の場合、元のリストを別のスレッドが書き換えていないか
  • 1つのタスクの失敗をどう扱うかが決まっているか。invokeAll()の後、各Futureのget()でExecutionExceptionを確かめているか

このチャプターで学んだこと

本チャプターでは、以下を取り上げました。

  1. Executorフレームワークが、タスクの投入と、スレッドプールによる実行を分離する仕組みについて
  2. スレッドプールの種類（単一、固定、動的、仮想スレッド）の動きと、上限の無い待ち行列・スレッド数の危険について
  3. shutdown()、shutdownNow()、awaitTermination()、close()によるシャットダウンの状態遷移と、割り込みに応じないタスクについて
  4. scheduleAtFixedRate()とscheduleWithFixedDelay()の違いと、定期実行のタスクが例外で止まる危険について
  5. FutureとCallableによる結果の受け取り、ExecutionExceptionによる例外の受け取り、get()のタイムアウトとcancel()について
  6. invokeAll()による一括実行と、共有せずに分割して集計する設計について
12
12
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
12
12

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?