java8 Gold 並列処理②

前(並列処理 Part１)

次(javaのクラス設計 Part1)

おはようございます！今回も、少しでも知識習得の助けになればと思います！

前回に引き続き、並列処理について触っていきましょう！

【対象者】本記事は、Oracle Certified Java Programmer, Gold SE 8 認定資格試験の出題分野の解説をいたします。特に以下のような方には読んでいただいて、理解の助けになればと思う次第です。

当該試験を受ける方
Gold SE 11の試験を受ける方で並列処理分野の理解を深めたい方
受験の有無にかかわらず、javaの並列処理についておさらいしたい方

出題範囲の確認

前回も掲載しましたが、

公式ページの出題内容は以下の通りです。

Java の同時実行性

Runnable と Callable を使用してワーカー・スレッドを作成する。 ExecutorService を使用してタスクを同時に実行する
スレッド化の潜在的な問題であるデッドロック、スタベーション、ライブロックおよび競合状態を識別する
キーワード synchronized と java.util.concurrent.atomic パッケージを使用してスレッドの実行順序を制御する
CyclicBarrier や CopyOnWriteArrayList など、java.util.concurrent のコレクションとクラスを使用する
並列 Fork/Join フレームワークを使用する
リダクション、分解、マージ・プロセス、パイプライン、パフォーマンスなど、並列ストリームを使用する

(https://education.oracle.com/ja/java-se-8-programmer-ii/pexam_1Z0-809)

これを読み替えて、、、

↓

１：Executorフレームワーク,ExecutorServiceとScheduledExecutorServiceの違い
１：Runnable、Callable型とそれぞれのタスクの性質の違い
２,３：スレッドセーフ系(atomic変数、java.util.concurrent.locks)の働き
３：syncronizedキーワード（関数宣言、シグニチャ記載箇所に書くもの）の意味とは性質
４：java.util.concurrentのクラス(CyclicBarrier や CopyOnWriteArrayList )の各々の特徴
５：fork/joinフレームワークって何をするものなのか
５：work-steelアルゴリズムって何
６：ストリーム（Stream api）の並列処理系メソッドはどう動くの

前回は前半、1,2,3,4について解説しました。今回は後半4,5,6,7について概要を説明するので、まずはふんわりとイメージを作っていきましょう！

それでは内容に入っていきます。

java.util.concurrentのクラス(CyclicBarrier とCopyOnWriteArrayList)の特徴

　前回の最初の方で述べましたが、java.util.concurrentパッケージはjava SE5で追加された並列処理ユーティリティです。並列処理の際のデータの整合性（操作の合間を縫って不正な処理が行われないか）や、処理を協調させるための機能があります。せっかく名前が出ているところだったので、以下のクラスから見ていきましょう。

ちなみにSilverでは名指しでクラス名が出るような場合はそのクラスのメソッドの概要や引数などを答える問題がある程度出易いようです(2020.12時点)。単純ですが試験対策に確認をおすすめします。

例 : Q… Predicateインタフェースのメソッド名は？ -> A… test() など

java.util.concurrent.CyclicBarrierクラス

概要

　スレッドパーティ(スレッドの集合のこと)の中で、各スレッドの進行具合を合わせる機能を提供します。「足並みを揃える」という表現が近いと思います。各スレッドは、ソース上に設定した「バリア」と呼ばれるコードブロック(実際には待機メソッドの行になります)に到達すると、他のスレッドがそのバリアに到達するまで待機します。

この待機状態は、所定のスレッドがすべてこのバリアに達すると実行を再開します。

実行再開を、「トリップ」と呼ぶ場合もあります。

まずは以下にコンストラクタの説明を書きます。波線はクラス名の略です。

-コンストラクタ-

~(int parties)
1. 「指定された数のスレッド(parties)が待機状態にある」という状況でトリップすることが可能なCyclicBarrierオブジェクトを作成します。バリアのトリップ時、定義済みアクションが実行されることはありません。
2. ちなみに、このpartiesの数ですが、当然ながらこの数がトリップの要件ですが、更に言うとスレッドパーティの中でトリップができるのはこの個数のスレッドのみです。
3. つまり、実際に動作するスレッド数よりもpartiesが少ない場合、partiesを超えた分のスレッドはいつまで経ってもトリップ出来なくなります。逆にpartiesが多い場合はいつまで経っても誰もトリップ自体ができません。
~(int parties, Runnable action)
1. 上記と同様のCyclicBarrierオブジェクトを作成します。しかし、こちらでは「そのバリアに最後に入ったスレッドが、actionの処理を実行する」というオマケ付きです。

-メソッド-

戻り値	メソッド名	動作
int	await()	全パーティーがバリアでawaitを呼ぶまで待機する。
int	int await(long timeout, TimeUnit unit)	全パーティーがバリアでawaitを呼ぶ　又はタイムアウト時間が経つまで待機する。
int	getNumberWaiting()	バリアで待機中のパーティー数を返す
int	getParties()	バリアの通過(トリップ)に必要なパーティーの数を返す
void	isBloken()	割り込みまたはタイムアウトによって破壊がされていないかを確認
void	reset()	バリアをリセット

【使い方】

メソッドの特徴から分かったとも思いますが、基本的には以下の流れで使用できます。
1. バリアオブジェクトを予測されるスレッド数に対して設定
2. 止まって欲しいところに<バリアオブジェクト>.await() を記述

CopyOnWriteArrayListクラス

概要

本クラスはArrayListの拡張版と思っていただいて大丈夫です。ArrayListとの違いは、リストの要素に変更を加えるadd,set,remobeといったメソッドの呼び出しに対して、内部で同じ要素を持った配列のコピーを作成する、という動作です。

ArrayListクラスとのメソッドの違い

“addAllAbsent”,”addIfAbsent”などがあり、逆に”remoceRange”といったメソッドは無くなっています。

他のConncurrentコレクションとの違い

　変更はコピー対象に、その瞬間に参照されても変更前の元配列を参照、といった動作のため、不整合が起こりにくくConcurrentModificationExceptionをスローすることはないという点が一番大きいかと思います。

利用のデメリット

　リストサイズが大きくなればなるほどコピー時の処理は重くなることが難点です。そのため、こちらよりsynchronizedListを使ったほうがいい、という意見もあります。

ただし、この辺りはプログラマによっても意見が違うところですので、
いざ自分が使うことになった時は目的とパフォーマンスから定量的に利用モジュールを比較する必要があると考えられます。

とまぁこんな感じで丸投げするのも感じが悪いので、
今後詳細記事を書いてそちらで解説できればと思います。

fork/joinフレームワーク

【概要】大きなタスクを小さなタスクに分割して、分割したタスクを複数スレッドで同時並行的に実行して処理パフォーマンスを上げるものです。つまり、マルチコアでのわかりやすい並列処理を行う仕組みです。

[クラス・インタフェース]

Fork/Joinフレームワークを利用するためのクラスは、以下の２つです。

ForkJoinPool
- ↓このタスクを実行する実装クラス。ExecutorService インタフェースをimplementsしています。

ForkJoinTask
- タスク本体となる中小クラス。↑によって実行されます。

-メソッド-

ForkJoinPoolクラスがForkJoinTaskを実行するメソッドは以下のとおりです。

void execute
- 非同期でタスクを実行。処理結果は受け取らず。
<T> T invoke
- タスクの実行が終了するまで待機し、処理結果を受け取る
<T> ForkJoinTask<T> submit
- 非同期でタスクを実行し、処理結果をタスクから受け取る。

[クラス・インタフェース] ~ForkJoinTaskクラスのサブクラス~

タスクのクラスを作る時には、以下のどちらかを継承します。

どちらも、computeという名前のメソッドをオーバーライドする必要があり、それを上記のexecuteなどのメソッドが呼ぶことで動作します。

RecursiveAction
- 処理結果として戻り値が入らない場合に使いましょう

RecursiveTask<V>
- 戻り値が欲しい場合はこちらを使いましょう<V>はcomputeの戻り値の型になります。

work-steelアルゴリズム

概要

上記のfork/joinフレームワークの中身となるもので、小分けにしたタスクを融通してやれる人がやる、といった考え方の効率化方法です。

複数のスレッドがワークキューを持っており、タスクがそれぞれに複数割り当てられたとします。

他のスレッドはスイスイタスクを片付けましたが、一人だけ手がついていないタスクがあるスレッドがいたとします。その場合、仕事が終わったスレッドがそのタスクキューの”後ろ”から持っていって自分のところで処理します。

後ろからタスクを取るため、先頭タスクを実行している元の所持者であるスレッドとの競合が起こることは”少なく”
タスクを終わらせたスレッドがまたもらい直すようなことも”少なく”、

といったように一応振り分けによる最適化はされやすくなっています。
(※ただ、競合やデータ不整合に関しては設計者の方で対策をしないとまずそうです)

ストリーム（Stream api）の並列処理系メソッドの動き

java8で追加されたstreamapiですが、これにも並列処理を利用するための機能があります。

ストリームの作成に用いるのは以下のメソッドですね。

メソッド名	動作
stream()	順次ストリームを生成
parallelStream()	並列ストリームを生成

この「並列 or 順次」　の状態に関しては、中間処理メソッドのparallel()とsequential()によって切り替えることもできます。

ただ、多くの情報源でも語られる通り、streamapiの並列処理機能を用いた時に、性能が必ず良くなるわけではありません。

並列ストリーム生成や確認に関するメソッドは以下のようなものがあります。

boolean isparallel()　並列ストリームであるかを確認する。

findAny()などのメソッドを並列ストリームで実行すると、処理されるタイミングもランダムとなり、それによっていつ終わるか、というのも処理内容によってばらつきが生じます。この辺りについては、今後コードベースで説明を追加できればと思います。

まとめ

　以上、java SE 8 Gold のための並列処理、第二回でした。
大まかなところを掴んで、まずは問題がある程度解けるようになるところからの方が入りやすいというのが私の感触です。

説明不足なところも多いかもしれませんが、
今後追加記事や加筆修正でよりわかりやすくしていければと思っております。

ご一読いただき、ありがとうございます！

前(並列処理 Part1)

次(javaのクラス設計 Part1)