Javaの並列/並行処理の基本

Yuichi.Sakuraba

June 29, 2023

Tweet Share

Transcript

1. Javaの並列/並行処理の基本 Java in the Box 櫻庭 祐一

2. Agenda Threadの基本 Concurrency Utilities 非同期タスクを記述する Concurrency Utilitiesの拡張 Folk/Join Framework CompletableFuture

3. Threadの基本

4. Thread 処理を並列/並行に処理するための最小単位 基本的にはOSスレッドのラッパー ただし、Virtual ThreadはJVMが管理するスレッド Threadに対する操作 OSスレッドに対する操作

5. Threadにおける2つの側面 ライフサイクルの管理 生成から廃棄まで スレッドスケジューリング スレッドの実行順序 実行中のスレッドの切り替え Context Switch

6. var thread = new Thread( new Runnable() { @Override public

   void run() { // 処理 while (!condition) { Thread.yield(); } // 処理 } }); thread.start(); スレッド生成 非同期タスク スレッドを譲る ただし、本当に譲るかどうかはスケジューラしだい 非同期タスクが完了でスレッド廃棄 タスクの実行 ただし、実際の実行タイミングはスケジューラが決める

7. スレッドの生成 時間がかかる メモリ大量消費 勝手に野良スレッドを作成されると管理が難しい コンテキストスイッチ 時間がかかる メモリ大量消費 勝手にスレッド切替されるとスケジューリングできない 非同期処理の実行・管理はJVMにまかせる 開発者は非同期タスクの記述に集中する

8. Thread を直接使用することは もはや アンチパターン 特に以下のメソッドは使用しない • stop() • suspend() •

   resume() これらのメソッドはJava 21から@Deprecated(forRemoval=true)

9. Concurrency Utilities

10. Concurrency Utilities 非同期タスクの実行・管理 並列コレクション アトミック操作 API ロック API

11. 非同期タスクの実行・管理 ExecutorService タスクの実行管理 主にスレッドプール Executors ExecutorServiceのファクトリ Runnable/Callable 非同期タスク Future 非同期タスクの管理

12. Executors ExecutorServiceのファクトリ 用途に合わせてExecutorServiceオブジェクトを生成 newSingleThreadExecutor newFixedThreadPool newCachedThreadPool newScheduledThreadPool newWorkStealingPool newVirtualThreadPerTaskExecutor シングルスレッド動作のExecutorServce

    スレッド数固定のスレッドプール 必要に応じてスレッド生成するスレッドプール 周期的タスクを実行するスレッドプール Work Stealingを使用するスレッドプール (Java 8) Virtual Threadを使用するExecutorService (Java 21)

13. ExecutorService 非同期タスクの実行 タスクの実行方法/スケジューリングは実装クラスに依存 主なメソッド submit close invokeAll/invokeAny shutdown/shutdownNow 非同期タスクの登録 戻り値はFuture<T>

    AutoClosable (Java 19) 複数タスクの実行 今後Structured Concurrencyで置き換え Java 19以前に使用していたExecutorServiceの終了

14. Runnable/Callable 非同期タスクを記述 @FunctionalInterfaceなので、ラムダ式で記述 Runnable Callable<T> 引数なし 戻り値なしのタスク Checked Exceptionはスローできない RuntimeExceptionはUncaught

    Exceptionとして扱われる 引数なし 戻り値ありのタスク Checked Exceptionをスローできる

15. Future<T> 非同期タスクの管理 型パラメータはタスクの戻り値型 Runnableの場合、Future<?> get cancel isDone/isCancelled resultNow exceptionNow state

    結果の取得 タスクが完了するまでブロックする タスクのキャンセル (キャンセルできるかはタスクしだい) タスクの状態を調べる ブロックせずに結果取得 完了していなければ例外 (Java 19) ブロックせずに例外取得 例外で完了していなければ例外 (Java 19) タスクの状態を調べる 戻り値はFuture.State列挙型 (Java 19)

16. final var path = ... try (var pool = Executors.newFixedThreadPool(

    Runtime.getRuntime().availableProcessors())) { Future<List<String>> future = pool.submit(() -> { return Files.readAllLines(path); }); List<String> contents = future.get(); } catch (ExecutionException ex) { // タスク実行時の例外 } catch (InterruptedException ex) { // タスク実行中に割り込み発生 } 例: 非同期ファイル読み込み 引数でスレッド数指定 ここではCPUのコア数を指定 タスク登録 IOExceptionが発生した場合 ExecutionExceptionのcauseとなり Future.getメソッドでcatchできる getはタスク完了までブロックするため このままだと並行処理の意味がない

17. final var path = ... try (var pool = Executors.newFixedThreadPool(

    Runtime.getRuntime().availableProcessors())) { Future<List<String>> future = pool.submit(() -> { return Files.readAllLines(path); }); /*たとえば*/ while(!future.isDone()) { /* タスク完了まで他の処理を実行 */ } List<String> contents = future.get(); } catch (ExecutionException ex) { // タスク実行時の例外 } catch (InterruptedException ex) { // タスク実行中に割り込み発生 } 例: 非同期ファイル読み込み 引数でスレッド数指定 ここではCPUのコア数を指定 タスク登録 IOExceptionが発生した場合 ExecutionExceptionのcauseとなり Future.getメソッドでcatchできる

18. 非同期タスクを記述する

19. 非同期タスクのポイント いわゆるスレッドセーフ 非同期実行でもデータを壊さない 実行タイミングによらず結果が同一 並行度と性能が比例する 並行処理のボトルネックが少 スケーラビリティ 安全性

20. 安全性を損なう要因 競り合い状態 複数スレッドが同タイミングで書き込みを行うことで データが破壊される データ書き込みの非可視化 メモリ書き込みタイミングにより予測不可能な動作になる

21. 競り合い状態の例 class Counter { private int count = 0; public

    int getNext() { return count++; } } read 複製 加算 write Thread #1 read 複製 加算 write Thread #2

22. 競り合い状態の例 class Counter { private int count = 0; public

    int getNext() { return count++; } } read 複製 加算 write Thread #1 read 複製 加算 write Thread #2 複数スレッドからcountにアクセス可能 アクセスを単一スレッドに制限 = 同期化

23. 競り合い状態の解消 class SyncCounter { private int count = 0; synchronized

    public int getNext() { return count++; } } class LockedCounter { private ReentrantLock lock = new ReentrantLock(); private int count = 0; public int getNext() { lock.lock(); try { return count++; } finally { lock.unlock(); } } }

24. class OneDataContainer { private int value; synchronized public int getValue()

    { return value; } synchronized public void setValue(int v) { value = v; } } boolean update(int value) { if (container.getValue() == value) { return false; } else { container.setValue(value); return true; } } スレッドセーフ？ Check Act Check-Then-Actは 処理全体を同期化する

25. class OneDataContainer { private int value; synchronized public int getValue()

    { return value; } synchronized public void setValue(int v) { value = v; } } synchronized boolean update(int value) { if (container.getValue() == value) { return false; } else { container.setValue(value); return true; } } スレッドセーフ？ synchronizedメソッドは synchronized(this) { … } containerが逸脱している場合 スレッドセーフではない

26. class OneDataContainer { private int value; synchronized public int getValue()

    { return value; } synchronized public void setValue(int v) { value = v; } } boolean update(int value) { synchronized(container) { if (container.getValue() == value) { return false; } else { container.setValue(value); return true; }} } containerを同期化することで スレッドセーフになる

27. データ書き込みの非可視化の例 public class NoVisibility { private static boolean ready; private

    static int number; public static void main(String... args) { try (var pool = Executors.newCachedThreadPool()){ pool.submit(() -> { while (!ready) { Thread.yield(); } System.out.println(number); }); number = 42; ready = true; } } } Java並行処理プログラミングp40より引用、改変 同期化により可視化を保証

28. 同期化 同期化を行うことで • リソースへのアクセスを単一スレッドに制限 複数スレッドでも逐次動作になりボトルネック化 • データ書き込みの可視化を保証 メインメモリへのアクセスにより多大な時間を消費 同期化することでスケーラビリティが低下

29. 安全性を担保しつつスケールさせるためには 同期化させる部分を最小限に 安易にsynchronizedメソッドを定義しない synchronizedよりもReentrantLockなどのロックAPIを使用 スレッド間でのリソースを共有しない リソースを共有するため競り合い状態などが発生する 引数、戻り値だけでスレッド間のやり取りを行う 状態を変更しない 状態が変更できるために競り合い状態などが発生する 状態を変更できないイミュータブル性を重視

    Java 16で導入されたRecord型を活用

30. スケーラビリティ向上させるために タスクの独立性 タスク間のやり取りやリソース共有を排除する タスクの均質化 同じタスクの並行処理は効率的 同じタスクでなくても、なるべく同質な処理にする タスク粒度の最適化 タスクの粒度が大きいとタスクスケジューリングが難しい 計算処理であれば細粒度 (分割統治法)

31. Concurrency Utilitiesの拡張

32. Concurrency Utilitiesの拡張 Fork/Join Framework (Java 7) 大量の計算処理、データ処理に対する非同期処理 応答性の向上 CompletableFuture (Java

    8) I/Oを含む業務ロジックの非同期処理 スループットの向上

33. Fork/Join Framework (Java 7) 大量の計算処理、データ処理を効率的に処理するフレームワーク キーとなる技術 分割統治法 Work-Stealingタスクスケジューリング Parallel Stream、Arrays.sortなどで使用

    オーバーヘッドがあるため、データ数が多い場合のみ Arrays.sortの場合、デフォルトで4096以上 開発者が直接使うことはほぼないはず…

34. 分割統治法 大きいタスクを処理しやすいサイズまで分割して処理 例: ソート 1 2 4 5 8 10

    11 3 1 2 4 5 8 10 11 3 1 2 4 5 8 10 11 3 4 10 1 5 11 2 8 3 2 10 1 5 11 4 8 3 8 11 5 2 4 10 3 1

35. Work-Stealingタスクスケジューリング 個々のスレッドがタスクキューを持つ タスクキューにタスクあり 先頭から取り出して処理 タスクキューにタスクなし 他スレッドのタスクキューの末尾からタスクを取り出す(Steal) 分割したタスクはキューに積まれる 末尾に近いほどタスクが小さい 効率的にタスクを処理できる

36. Fork/Join FrameworkのAPI ForkJoinPool Work-Stealを使用したスレッドプール ForkJoinTask 分割統治法を使用してタスクを記述する基底クラス RecursiveTask/RecursiveAction 再帰的タスクを記述用クラス RecursiveActionは戻り値なし CountedCompleter

    保留中のタスクがない場合に完了アクションを記述できるタスク

37. 例: フィボナッチ数 F(n) = F(n-1) + F(n-2) class FibonacciTask extends

    RecursiveTask<Integer> { private final int n; public FibonacciTask(int n) { this.n = n; } protected Integer compute() { if (n <= 1) return n; var f1 = new FibonacciTask(n - 1); f1.fork(); var f2 = new FibonacciTask(n - 2); return f2.compute() + f1.join(); }} ForkJoinPool fjPool = new ForkJoinPool(); var task = fjPool.submit(new FibonacciTask(30)); System.out.println(task.get()); タスク分割 タスク分割 タスクのフォーク タスクの完了を待って、結果を取得 タスク記述メソッド

38. CompletableFuture (Java 8) 非同期で行う一連の処理を関数で連ねる I/O処理を非同期で行うことでスループット向上 開始 (staticメソッド) completedFuture runAsync suplyAsync

    一連のタスク記述メソッド thenAccept/Apply/Run(Async) 他に処理の合成、例外処理など多くのメソッドを提供 引数が次のラムダ式の引数になる タスクはRunnable タスクはSupplier 引数、戻り値の有無でラムダ式が決まる メソッドの最後がAsyncの場合、非同期に実行される

39. CompletableFuture.supplyAsync(() -> Path.of(...)) .thenApplyAsync(path -> { try { return Optional.of(Files.readAllLines(path));

    } catch (IOException ex) { return Optional.<List<String>>empty(); } }) .thenAccept(opt -> { opt.ifPresent(contents -> { contents.forEach(System.out::println); }); }); 例: 非同期ファイル読み込み 引数あり、戻り値ありのタスクを非同期に実行 ファイル読み込みタスク完了後、実行

40. Conclusion Threadを直接使用するのはもはやアンチパターン Concurrency Utilitiesでスレッド管理とタスクを分離 安全性とスケーラビリティに注意してタスクを記述 応答性向上: Fork/Join Framework スループット向上: CompletableFuture

41. おまけ Javaで並列・並行処理を行うのであれば 絶版だけど… 電子版もないけど…

42. Javaの並列/並行処理の基本 Java in the Box 櫻庭 祐一