Jakarta Concurrencyによる並行処理プログラミングの始め方 (JJUG CCC 2024 Fall)

Transcript

1. © 2024 Fujitsu Limited Jakarta Concurrencyによる 並行処理プログラミングの始め方 2024/10/27 長尾 貴浩 JJUG CCC 2024 Fall #jjug_ccc #jjjug_ccc_l
2. 自己紹介 © 2024 Fujitsu Limited 2 ⚫ 富士通株式会社 ソフトウェアエンジニア ⚫ アプリケーションサーバー製品の開発・保守 ⚫ Jakarta EE/MicroProfile関連のOSS開発 ⚫ MicroProfileステコミ委員 ⚫ X (Twitter): @tnagao7 ⚫ JJUG CCC 2019 SpringのLTで登壇 長尾 貴浩
3. 目次 ⚫Jakarta Concurrencyとは何か？ ⚫Jakarta Concurrencyによる並行処理プログラミング ⚫Java SEのConcurrency Utilitiesとの関係 ⚫Jakarta Concurrencyが提供する4つの管理オブジェクト ⚫Jakarta Concurrency 3.1の強化ポイント © 2024 Fujitsu Limited 3
4. Javaの並行処理とスレッド © 2024 Fujitsu Limited 4 メインスレッド 処理1() 処理2() t メインスレッド スレッド1 処理1() スレッド2 処理2() Javaのプロセス内の並行処理にはスレッドが使われる
5. Jakarta EEアプリでスレッドを使えるか？ (1/2) © 2024 Fujitsu Limited 5 リクエスト 処理スレッド thread1 post() thread2 @Path("/sample") public class SampleResource { @POST public void post() { Thread thread1 = new Thread(() -> { 処理1(); }); thread1.start(); // スレッド起動 Thread thread2 = new Thread(() -> { 処理2(); }); thread2.start(); // スレッド起動 } } thread1 thread2 Jakarta EEアプリケーションでもスレッドを使いたい！ でも...
6. Jakarta EEアプリでスレッドを使えるか？ (2/2) © 2024 Fujitsu Limited 6 リクエスト 処理スレッド thread1 post() thread2 @Path("/sample") public class SampleResource { @POST public void post() { Thread thread1 = new Thread(() -> { 処理1(); }); thread1.start(); // スレッド起動 Thread thread2 = new Thread(() -> { 処理2(); }); thread2.start(); // スレッド起動 } } thread1 thread2 new Thread() で作ったスレッドが正常に動かないケースがある... javax.naming.NamingException: Lookup failed for java:comp/env/jdbc/ds 例外発生 実際の例外の例
7. Jakarta EEにおけるスレッド利用時の問題 (1/2) © 2024 Fujitsu Limited 7 ⚫Jakarta EEでは自前のスレッド作成が推奨されていない The enterprise bean must not attempt to manage threads. The enterprise bean must not attempt to start, stop, suspend, or resume a thread, or to change a thread’s priority or name. ... (日本語訳: Enterprise Beanは、スレッドを管理しようとしてはならない。 Enterprise Beanは、スレッドを開始、停止、中断、再開しようとしたり、 スレッドの優先度や名前を変更しようとしてはならない。 ...) ― Jakarta Enterprise Beans 4.0規約より https://jakarta.ee/ja/specifications/enterprise-beans/4.0/
8. Jakarta EEにおけるスレッド利用時の問題 (2/2) © 2024 Fujitsu Limited 8 コンテキスト {java:comp/env/jdbc/ds= DataSouce@12345678, ...} 非管理スレッド コンテキスト { } Java SE APIで スレッドを作成 ⚫Jakarta EEのスレッドはコンテキスト(環境情報) をもつ ⚫Jakarta EEの機能はスレッドごとのコンテキストをもとに動作する ⚫new Thread() だとJakarta EEのコンテキストが引き継がれない リクエスト処理スレッド 実行中アプリケーションの環境情報 (JNDI, クラスローダー, セキュリティ, トランザクション, ...) コンテキストが引き継がれず Jakarta EEの機能を利用すると エラーが発生してしまう...
9. Jakarta Concurrency © 2024 Fujitsu Limited 9 コンテキスト {java:comp/env/jdbc/ds= DataSouce@12345678, ...} 管理スレッド Jakarta Concurrency を利用 ⚫Jakarta EEコンテナによって管理された方法で並行処理を実行 ⚫Java SEのConcurrency Utilities (JSR 166) と同じインタフェース ⚫管理スレッド(コンテキスト付きのスレッド) が問題を解決 リクエスト処理スレッド コンテキスト {java:comp/env/jdbc/ds= DataSouce@12345678, ...} Jakarta EEの機能を 正常に利用できる
10. 目次 © 2024 Fujitsu Limited 10 ⚫Jakarta Concurrencyとは何か？ ⚫Jakarta Concurrencyによる並行処理プログラミング ⚫Java SEのConcurrency Utilitiesとの関係 ⚫Jakarta Concurrencyが提供する4つの管理オブジェクト ⚫Jakarta Concurrency 3.1の強化ポイント
11. Jakarta EE © 2024 Fujitsu Limited 11 [例] データベースを利用するWebアプリケーションのシステム構成 Java VM Jakarta EE コンテナ (アプリケーションサーバー) Jakarta EE アプリケーション データベース HTTP クライアント ⚫Javaによるエンタープライズシステム向け規約群 ⚫2024年にJakarta EE 11がリリース予定 ⚫アプリケーションサーバーのスレッド上でアプリが動作 ⚫アプリケーションサーバー自身がマルチスレッドプログラム
12. Jakarta EEにおけるスレッド利用 © 2024 Fujitsu Limited 12 リクエスト 処理スレッド スレッド1 service() run() スレッド2 run() リクエスト 受付スレッド リクエスト 処理スレッド アプリケーションサーバー アプリケーション
13. Jakarta EEとJakarta Concurrency © 2024 Fujitsu Limited 13 https://jakarta.ee/specifications/concurrency/3.1/jakarta-concurrency-spec-3.1
14. Java SE/Jakarta EEの並行処理ユーティリティ © 2024 Fujitsu Limited 14 Java 5 JSR 166 Concurrency Utilities Java SE Java EE 7 JSR 236 Concurrency Utilities for Java EE 1.0 Jakarta EE 9 Jakarta Concurrency 2.0 (javax→jakarta) Jakarta EE 10 Jakarta EE 8 Jakarta Concurrency 1.1 (財団への移管) Jakarta Concurrency 3.0 (機能追加) Jakarta EE
15. Jakarta Concurrencyの管理オブジェクト © 2024 Fujitsu Limited 15 Executor Service java.util.concurrent Scheduled Executor Service Thread Factory Jakarta Concurrencyでは4種類の管理オブジェクトが利用できる Concurrency Utilities (Java SE) Jakarta Concurrency Managed Executor Service Context Service jakarta.enterprise.concurrent Managed Scheduled Executor Service Managed Thread Factory
16. ExecutorServiceとは © 2024 Fujitsu Limited 16 スレッド1 スレッド2 スレッド3 スレッド4 スレッド5 タスク 2 スレッドプール キュー [例] ThreadPoolExecutor (ExecutorServiceの実装クラス例) Java SE 実行したタスクの進捗の追跡が可能な非同期処理実行サービス タスク 1 public class SampleProgram { public static final int POOL_SIZE = 5; public static void main(String[] args) { ExecutorService executorSvc = Executors.newFixedThreadPool(POOL_SIZE); executorSvc.submit(() -> { // タスク1 }); executorSvc.submit(() -> { // タスク2 }); ... }
17. ManagedExecutorServiceの使用例 © 2024 Fujitsu Limited 17 Java SE Jakarta EE public class SampleProgram { public static final int POOL_SIZE = 5; public static void main(String[] args) { ExecutorService executorSvc = Executors .newFixedThreadPool(POOL_SIZE); executorSvc.submit(() -> { // タスク1 }); executorSvc.submit(() -> { // タスク2 }); ... } } @Path("/sample") public class SampleResource { @Resource ManagedExecutorService executorSvc; @POST public void post() { executorSvc.submit(() -> { // タスク1 }); executorSvc.submit(() -> { // タスク2 }); ... } }
18. ManagedScheduledExecutorService © 2024 Fujitsu Limited 18 Java SE Jakarta EE タイマー付きの非同期処理実行サービス 10秒後に開始, 5秒間隔で繰り返し @Path("/sample") public class SampleResource { @Resource ManagedScheduledExecutorService scheduler; @POST public void post() { scheduler .scheduleWithFixedDelay(() -> { // タスク1 }, 10, 5, TimeUnit.SECONDS); ... } } public class SampleProgram { public static final int POOL_SIZE = 5; public static void main(String[] args) { ScheduledExecutorService scheduler = Executors. newScheduledThreadPool(POOL_SIZE); scheduler .scheduleWithFixedDelay(() -> { // タスク1 }, 10, 5, TimeUnit.SECONDS); ... } }
19. ManagedThreadFactory © 2024 Fujitsu Limited 19 Jakarta EE スレッド作成用のファクトリクラス newThread() で 新しいスレッドを作成 @Path("/sample") public class SampleResource { @Resource ManagedThreadFactory threadFactory; @POST public void post() { Thread thread1 = threadFactory.newThread(() -> { // タスク1 }); thread1.start(); ... } } Java SE public class SampleProgram { public static void main(String[] args) { ThreadFactory threadFactory = Executors.defaultThreadFactory(); Thread thread1 = threadFactory.newThread(() -> { // タスク1 }); thread1.start(); ... } }
20. ContextService © 2024 Fujitsu Limited 20 コンテキスト 付与 コンテキスト タスク コンテキスト付きタスク (Contextual Task) タスク タスクにコンテキストを 付与するサービス Jakarta EE @ContextServiceDefinition( name = "java:module/concurrent/CustomContext", propagated = { SECURITY, APPLICATION }, ...) @ManagedExecutorDefinition( name = "java:module/concurrent/CustomExecutor", context = "java:module/concurrent/CustomContext) @Path("/sample") public class SampleResource { @Resource(lookup = "java:module/concurrent/CustomExecutor") ManagedExecutorService executorService; @POST public void post() { executorService.submit(() -> { ... }); } }
21. Jakarta EE並行処理のその他の選択肢 Jakarta EE/MicroProfileで利用できる並行処理機能は他にもある ⚫Jakarta Concurrency ⚫非同期メソッド (@jakarta.enterprise.context.Asynchronous) ⚫Jakarta Servlet ⚫非同期サーブレット ⚫Jakarta Enterprise Beans ⚫非同期メソッド (@jakarta.ejb.Asynchronous) ⚫タイマーサービス ⚫MicroProfile ⚫MicroProfile Fault Tolerance (@org.eclipse.microprofile.faulttolerance.Asynchronous) ⚫MicroProfile Reactive Messaging, Reactive Streams Operators ⚫MicroProfile Context Propagation © 2024 Fujitsu Limited 21
22. 目次 © 2024 Fujitsu Limited 22 ⚫Jakarta Concurrencyとは何か？ ⚫Jakarta Concurrencyによる並行処理プログラミング ⚫Java SEのConcurrency Utilitiesとの関係 ⚫Jakarta Concurrencyが提供する4つの管理オブジェクト ⚫Jakarta Concurrency 3.1の強化ポイント
23. Jakarta Concurrency 3.1が対応強化するJDK機能 © 2024 Fujitsu Limited 23 JDK 5 JSR 166 Concurrency Utilities JDK Jakarta EE Jakarta EE 11 Jakarta Concurrency 3.1 JDK 9 JEP 266 Flow API JDK 21 JEP 444 仮想スレッド 2024年リリース予定
24. Jakarta Concurrency 3.1の強化ポイント ⚫Flow API対応強化 ⚫Flow API (Reactive Streams) のコンテキスト対応 ⚫仮想スレッド対応 ⚫並行処理の実行スレッドとして仮想スレッドが使用可能に ⚫Jakarta Enterprise Beansの一部機能の移植 ⚫定期実行処理のための@Scheduleアノテーションの導入 ⚫その他 ⚫@Resourceの代わりに@Injectが使用可能に © 2024 Fujitsu Limited 24
25. JDKのFlow API © 2024 Fujitsu Limited 25 ⚫Reactive Streams: 非同期ストリーム処理の仕様の一つ ⚫asynchronous stream processing with non-blocking back pressure (*) ⚫java.util.concurrent.Flowクラス内に Reactive Streamsのインタフェースが宣言されている Flow. Publisher onNext(data) request(n) Flow. Subscription Flow. Subscriber SubmissionPublisherが スレッドプールをもつ (*) https://www.reactive-streams.org
26. Jakarta ConcurrencyのFlow API対応強化 © 2024 Fujitsu Limited 26 ⚫Flow APIにおけるコンテキスト付与の仕組みが整備された ⚫Executorを変更できないPublisherでもコンテキストを考慮できるようになった Executorを変更可能なPublisherであれば 代わりにManagedExecutorServiceを使ってもいい 参考 publisher.subscribe( contextService.contextualSubscriber(subscriber)); Publisherのスレッドで呼び出される Subscriberにコンテキストを付与
27. OpenJDK 21で正式導入された仮想スレッド © 2024 Fujitsu Limited 27 ⚫Java VMが提供する軽量なスレッド ⚫従来のスレッド (プラットフォームスレッド) はOSのスレッドと1 : 1で紐づく ⚫仮想スレッドは内部的にプラットフォームスレッド にマウントされて動作する (M : N) キャリアスレッド1 (プラットフォームスレッド) 仮想 スレッド2 仮想 スレッド1 I/O処理 I/O待ちが発生する 並行処理で プラットフォームスレッドを 有効活用できる
28. Jakarta Concurrencyの仮想スレッド対応 (1/2) © 2024 Fujitsu Limited 28 ⚫並行処理の実行スレッドとして仮想スレッドを選択できる ⚫アノテーションやデプロイメントディスクリプタ (application.xmlなど) で設定可能 ⚫動作条件 ⚫OpenJDK 21以上 ⚫Jakarta Concurrencyの仮想スレッド実行に対応したアプリケーションサーバー I/Oバウンドな並行処理でスループット向上が期待できる 仮想スレッドが常に最適解とは限らない―効果を評価してから選択を ⚫ CPUバウンドな処理では仮想スレッドが性能向上に貢献しない ⚫ ThreadLocal変数の使われ方次第では性能が劣化するかも (ThreadLocalを高コストオブジェクトのキャッシュとする使い方は相性が悪い)
29. Jakarta Concurrencyの仮想スレッド対応 (2/2) © 2024 Fujitsu Limited 29 @ManagedExecutorDefinition( name = "java:module/concurrent/VExecutor", virtual = true) @Path("/sample") public class SampleResource { @Resource(lookup = "java:app/concurrent/VExecutor") ManagedExecutorService executorSvc; @POST public void post() { executorSvc.submit(() -> { // タスク }); ... } } <managed-executor> <name>java:module/concurrent/VExecutor</name> <virtual>true</virtual> </managed-executor> 定義ファイル (デプロイメントディスクリプタ) を使う場合 virtual = trueで仮想スレッドを有効化
30. @Scheduleアノテーションの追加 © 2024 Fujitsu Limited 30 ⚫@Scheduleアノテーションで定期実行処理が記述可能に ⚫Jakarta Enterprise Beansのタイマーサービスの後継 import jakarta.enterprise.concurrent.Asynchronous; import jakarta.enterprise.concurrent.Schedule; @ApplicationScoped public class SampleBean { @Asynchronous(runAt = @Schedule(hours = { 0 }, ...)) public void scheduledTask() { ... } } @Asynchronous(runAt = @Schedule(cron = "*/5 * * * * *")) public void scheduledTask() { ... } 5秒ごとに実行 0時に実行
31. まとめ © 2024 Fujitsu Limited 31 Jakarta Concurrencyはコンテキスト付きスレッドを提供 スレッドからJakarta EEの機能が利用できない問題を解決 Java SEのConcurrency Utilitiesと同じ使い方ができる Java SEプログラムと同じインタフェースで並行処理を実装できる Jakarta EE 11 (Jakarta Concurrency 3.1) で JDKの並行処理機能をもっと活用できる 仮想スレッドによってI/Oバウンドな並行処理のスループットを向上
32. Thank you © 2024 Fujitsu Limited