[Java8]ParallelStream内で例外が発生した時スレッドは終了するのか？

どうも、トムです。

そろそろ世間(?)でも徐々に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点

ParallelStream内のスレッドでは親スレッドと同じスレッドも使用する…らしい

ParallelStream内で例外が発生しても正常処理のスレッドは継続する…らしい

ParallelStreamでの同時実行可能スレッド数は5スレッド…らしい

※同時実行可能スレッド数に関してはJVMの設定や環境によるかもしれません

う～ん……ここら辺はもっと検証してみないと何とも言えないですね。
ただ、親スレッドと同じスレッドで並列処理が走るのは何とも奇妙な……

サーバサイドアプリでの検証

念のためサーバサイドで実装した場合にも同じ動きをするのか検証してみましょう。

今回検証に利用したのはPlay Framework 2.2.2
Play Frameworkは、J2EEを完全に排除し複雑なサーバサイドの構築やらTomcatのようなサーブレットコンテナを必要とせずにRubyのRailsやPHPのCakePHPのようにお手軽にMVCアーキテクチャを利用したアプリを作成できる生産性を重視するWebアプリ開発に特化した次世代フレームワークです。

Play Frameworkの構築記事はまた次の機会という事で、さっそく検証してみましょう。


何回か実行してみます。

～～～(中略)～～～

サーバサイドでも特に動きに変わりはないようです。
スレッド数に関しては安心ですが、ParallelStreamの例外時には正常処理のスレッドが残ってしまうので気を付けた方がよさそうです。
やる人はいないと思いますがParallelStreamでDB操作とかは絶対やっちゃダメですね。