そろそろ世間(?)でも徐々にJava SE8の話題が増えてきた今日この頃
Web上を彷徨っているとチラッとこんな記事↓を見かけました。
もしもラムダの中で例外が発生したら（前編）
この記事の中でParallelStreamについての検証があるのですが
発生した例外は誰が受け取るのか？どこまで伝播するのか？すごく気になりますね。
例外が発生したスレッド全てでcatch処理が行われるのか？
例外が発生しなかった他のスレッドはどのような影響を受けるのか？
そもそもちゃんと全部のスレッド処理が終了するのか？
皆さんも気になりますよね？
ParallelStream例外時のスレッドの動きを観察
引用記事の中ではParallelStream内で例外が発生した場合の結果として幾つかのスレッドが残ったままになっているように見えますが
例外が発生するたびスレッドがどんどん増え続けるなんて事にならないのかちょっと確かめてみます。
コードは引用記事の内容を拝借しました。
parallelStreamで並列処理中に"ERROR"という文字列が出てきたらRuntimeExceptionをthrowするという処理を10回繰り返しスレッド数の変化を確認します。
List<String> lines = Arrays.asList("A", "B", "C", "D", "ERROR", "F", "G", "ERROR", "I", "J");
// 取り敢えず10回繰り返して様子見
for (int i=0; i<10; i++) {
System.out.println("スレッド数： " + String.valueOf(Thread.activeCount()));
System.out.println("親スレッド Thread=" + Thread.currentThread().getId());
Set<String> threads = new CopyOnWriteArraySet<>();
try {
lines.parallelStream().forEach(s -> {
System.out.println("=>開始[" + Thread.currentThread().getId() + "] 文字[" + s + "]" + ", スレッド数=" + String.valueOf(Thread.activeCount()));
threads.add(String.valueOf(Thread.currentThread().getId()));
try {
Thread.sleep(100L);
if (s.equals("ERROR")) {
System.out.println("◇◇◇◇ラムダ内で例外発生◇◇◇◇ Thread=" + Thread.currentThread().getId() + ", スレッド数=" + String.valueOf(Thread.activeCount()));
System.out.println("<=エラー終了[" + Thread.currentThread().getId() + "] 文字[" + s + "]" + ", スレッド数=" + String.valueOf(Thread.activeCount()));
threads.remove(String.valueOf(Thread.currentThread().getId()));
throw new RuntimeException();
}
} catch (InterruptedException e) {
System.out.println("InterruptedException in lambda. Thread=" + Thread.currentThread().getId() + ", スレッド数=" + String.valueOf(Thread.activeCount()));
}
System.out.println("<=終了[" + Thread.currentThread().getId() + "] 文字[" + s + "]" + ", スレッド数=" + String.valueOf(Thread.activeCount()));
threads.remove(String.valueOf(Thread.currentThread().getId()));
});
} catch (Exception ex) {
System.out.println("◆◆◆◆ラムダの外で例外キャッチ◆◆◆◆ Thread=" + Thread.currentThread().getId() + ", スレッド数=" + String.valueOf(Thread.activeCount()));
System.out.println("ラムダ内で終わっていないように見えるスレッド： [" + String.join(",", threads) + "]" + ", スレッド数=" + String.valueOf(Thread.activeCount()));
}
System.out.println("スレッド数： " + String.valueOf(Thread.activeCount()));
System.out.println("----------------------------------------------------------------------");
}
動かしてみると………
スレッド数： 2 親スレッド Thread=1 =>開始[1] 文字[G], スレッド数=5 =>開始[12] 文字[I], スレッド数=5 =>開始[11] 文字[C], スレッド数=5 =>開始[13] 文字[ERROR], スレッド数=5 <=終了[1] 文字[G], スレッド数=5 <=終了[11] 文字[C], スレッド数=5 =>開始[11] 文字[ERROR], スレッド数=5 <=終了[12] 文字[I], スレッド数=5 =>開始[12] 文字[J], スレッド数=5 ◇◇◇◇ラムダ内で例外発生◇◇◇◇ Thread=13, スレッド数=5 <=エラー終了[13] 文字[ERROR], スレッド数=5 =>開始[13] 文字[B], スレッド数=5 =>開始[1] 文字[F], スレッド数=5 ◇◇◇◇ラムダ内で例外発生◇◇◇◇ Thread=11, スレッド数=5 <=エラー終了[11] 文字[ERROR], スレッド数=5 =>開始[11] 文字[D], スレッド数=5 <=終了[1] 文字[F], スレッド数=5 ◆◆◆◆ラムダの外で例外キャッチ◆◆◆◆ Thread=1, スレッド数=5 ラムダ内で終わっていないように見えるスレッド： [12,13,11], スレッド数=5 スレッド数： 5 ---------------------------------------------------------------------- <=終了[12] 文字[J], スレッド数=5 <=終了[13] 文字[B], スレッド数=5 <=終了[11] 文字[D], スレッド数=5 【↑ここまで前回のラムダ処理のスレッド↑】 ※この部分だけ順番入れ替えています スレッド数： 5 親スレッド Thread=1 =>開始[12] 文字[A], スレッド数=5 =>開始[1] 文字[G], スレッド数=5 =>開始[13] 文字[C], スレッド数=5 =>開始[11] 文字[B], スレッド数=5 <=終了[1] 文字[G], スレッド数=5 =>開始[1] 文字[F], スレッド数=5 <=終了[13] 文字[C], スレッド数=5 =>開始[13] 文字[ERROR], スレッド数=5 <=終了[12] 文字[A], スレッド数=5 =>開始[12] 文字[D], スレッド数=5 <=終了[11] 文字[B], スレッド数=5 =>開始[11] 文字[A], スレッド数=5 ◇◇◇◇ラムダ内で例外発生◇◇◇◇ Thread=13, スレッド数=5 <=エラー終了[13] 文字[ERROR], スレッド数=5 <=終了[1] 文字[F], スレッド数=5 =>開始[1] 文字[I], スレッド数=5 =>開始[13] 文字[ERROR], スレッド数=5 <=終了[12] 文字[D], スレッド数=5 <=終了[11] 文字[A], スレッド数=5 <=終了[1] 文字[I], スレッド数=5 ◆◆◆◆ラムダの外で例外キャッチ◆◆◆◆ Thread=1, スレッド数=5 ラムダ内で終わっていないように見えるスレッド： [13], スレッド数=5 スレッド数： 5 ---------------------------------------------------------------------- ◇◇◇◇ラムダ内で例外発生◇◇◇◇ Thread=13, スレッド数=5 <=エラー終了[13] 文字[ERROR], スレッド数=5 【↑ここまで前回のラムダ処理のスレッド↑】 ※この部分だけ順番入れ替えています スレッド数： 5 親スレッド Thread=1 =>開始[11] 文字[C], スレッド数=5 =>開始[12] 文字[B], スレッド数=5 =>開始[1] 文字[G], スレッド数=5 =>開始[13] 文字[ERROR], スレッド数=5 <=終了[11] 文字[C], スレッド数=5 =>開始[11] 文字[I], スレッド数=5 <=終了[12] 文字[B], スレッド数=5 =>開始[12] 文字[A], スレッド数=5 <=終了[1] 文字[G], スレッド数=5 =>開始[1] 文字[F], スレッド数=5 ◇◇◇◇ラムダ内で例外発生◇◇◇◇ Thread=13, スレッド数=5 <=エラー終了[13] 文字[ERROR], スレッド数=5 =>開始[13] 文字[D], スレッド数=5 <=終了[12] 文字[A], スレッド数=5 =>開始[12] 文字[ERROR], スレッド数=5 <=終了[11] 文字[I], スレッド数=5 <=終了[1] 文字[F], スレッド数=5 ◆◆◆◆ラムダの外で例外キャッチ◆◆◆◆ Thread=1, スレッド数=5 ラムダ内で終わっていないように見えるスレッド： [13,12], スレッド数=5 スレッド数： 5 ---------------------------------------------------------------------- ～～～(中略)～～～ スレッド数： 5 ----------------------------------------------------------------------はい、スレッドは増えてはいないですね。
例外でスレッド数がもりもり増える事はないようです。
他に、気になる点は3点
う～ん……ここら辺はもっと検証してみないと何とも言えないですね。
ただ、親スレッドと同じスレッドで並列処理が走るのは何とも奇妙な……
サーバサイドアプリでの検証
念のためサーバサイドで実装した場合にも同じ動きをするのか検証してみましょう。今回検証に利用したのはPlay Framework 2.2.2
Play Frameworkは、J2EEを完全に排除し複雑なサーバサイドの構築やらTomcatのようなサーブレットコンテナを必要とせずにRubyのRailsやPHPのCakePHPのようにお手軽にMVCアーキテクチャを利用したアプリを作成できる生産性を重視するWebアプリ開発に特化した次世代フレームワークです。
Play Frameworkの構築記事はまた次の機会という事で、さっそく検証してみましょう。
何回か実行してみます。
～～～(中略)～～～
サーバサイドでも特に動きに変わりはないようです。
スレッド数に関しては安心ですが、ParallelStreamの例外時には正常処理のスレッドが残ってしまうので気を付けた方がよさそうです。
やる人はいないと思いますがParallelStreamでDB操作とかは絶対やっちゃダメですね。
