Thread.ofVirtual()メソッドをコールして、仮想スレッドを作成するためのThread.Builderのインスタンスを作成します。

次の例では、メッセージを出力する仮想スレッドを作成して開始します。これはjoinメソッドをコールして、仮想スレッドが終了するまで待機します。(これにより、出力れたメッセージをメイン・スレッドが終了する前に確認できます。)

Thread thread = Thread.ofVirtual().start(() -> System.out.println("Hello"));
thread.join();

Thread.Builderインタフェースを使用すると、一般的なThreadプロパティ(スレッドの名前など)のスレッドを作成できます。Thread.Builder.OfPlatformサブインタフェースによってプラットフォーム・スレッドが作成され、Thread.Builder.OfVirtualによって仮想スレッドが作成されます。

次の例では、Thread.Builderインタフェースを使用してMyThreadという名前の仮想スレッドを作成します:

Thread.Builder builder = Thread.ofVirtual().name("MyThread");
Runnable task = () -> {
    System.out.println("Running thread");
};
Thread t = builder.start(task);
System.out.println("Thread t name: " + t.getName());
t.join();

次の例では、Thread.Builderを使用して2つの仮想スレッドを作成して開始します:

Thread.Builder builder = Thread.ofVirtual().name("worker-", 0);
Runnable task = () -> {
    System.out.println("Thread ID: " + Thread.currentThread().threadId());
};

// name "worker-0"
Thread t1 = builder.start(task);   
t1.join();
System.out.println(t1.getName() + " terminated");

// name "worker-1"
Thread t2 = builder.start(task);   
t2.join();  
System.out.println(t2.getName() + " terminated");

この例では、次のような出力が表示されます:

Thread ID: 21
worker-0 terminated
Thread ID: 24
worker-1 terminated

エグゼキュータを使用すると、アプリケーションの他の部分からスレッド管理と作成を分離できます。

次の例では、Executors.newVirtualThreadPerTaskExecutor()メソッドを使用してExecutorServiceを作成します。ExecutorService.submit(Runnable)がコールされるたびに、新しい仮想スレッドが作成され、タスクを実行するために開始されます。このメソッドは、Futureのインスタンスを戻します。メソッドFuture.get()は、スレッドのタスクが完了するまで待機します。したがって、この例では、仮想スレッドのタスクが完了したときにメッセージが出力されます。

try (ExecutorService myExecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    Future<?> future = myExecutor.submit(() -> System.out.println("Running thread"));
    future.get();
    System.out.println("Task completed");
    // ...

次の例は、2つのクラスで構成されています。EchoServerは、ポートでリスニングし、接続ごとに新しい仮想スレッドを開始するサーバー・プログラムです。EchoClientは、サーバーに接続し、コマンドラインに入力されたメッセージを送信するクライアント・プログラムです。

EchoClientはソケットを作成し、EchoServerへの接続を取得します。これは標準入力ストリームでユーザーからの入力を読み取り、そのテキストをソケットに書き込むことでEchoServerに転送します。EchoServerは、ソケットを介して、その入力をEchoClientにエコーします。EchoClientは、サーバーから戻されたデータを読み取って表示します。EchoServerは、仮想スレッド(クライアント接続ごとに1つのスレッド)を介して複数のクライアントに同時にサービスを提供できます。

public class EchoServer {
    
    public static void main(String[] args) throws IOException {
         
        if (args.length != 1) {
            System.err.println("Usage: java EchoServer <port>");
            System.exit(1);
        }
         
        int portNumber = Integer.parseInt(args[0]);
        try (
            ServerSocket serverSocket =
                new ServerSocket(Integer.parseInt(args[0]));
        ) {                
            while (true) {
                Socket clientSocket = serverSocket.accept();
                // Accept incoming connections
                // Start a service thread
                Thread.ofVirtual().start(() -> {
                    try (
                        PrintWriter out =
                            new PrintWriter(clientSocket.getOutputStream(), true);
                        BufferedReader in = new BufferedReader(
                            new InputStreamReader(clientSocket.getInputStream()));
                    ) {
                        String inputLine;
                        while ((inputLine = in.readLine()) != null) {
                            System.out.println(inputLine);
                            out.println(inputLine);
                        }
                    
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                });
            }
        } catch (IOException e) {
            System.out.println("Exception caught when trying to listen on port "
                + portNumber + " or listening for a connection");
            System.out.println(e.getMessage());
        }
    }
}

public class EchoClient {
    public static void main(String[] args) throws IOException {
        if (args.length != 2) {
            System.err.println(
                "Usage: java EchoClient <hostname> <port>");
            System.exit(1);
        }
        String hostName = args[0];
        int portNumber = Integer.parseInt(args[1]);
        try (
            Socket echoSocket = new Socket(hostName, portNumber);
            PrintWriter out =
                new PrintWriter(echoSocket.getOutputStream(), true);
            BufferedReader in =
                new BufferedReader(
                    new InputStreamReader(echoSocket.getInputStream()));
        ) {
            BufferedReader stdIn =
                new BufferedReader(
                    new InputStreamReader(System.in));
            String userInput;
            while ((userInput = stdIn.readLine()) != null) {
                out.println(userInput);
                System.out.println("echo: " + in.readLine());
                if (userInput.equals("bye")) break;
            }
        } catch (UnknownHostException e) {
            System.err.println("Don't know about host " + hostName);
            System.exit(1);
        } catch (IOException e) {
            System.err.println("Couldn't get I/O for the connection to " +
                hostName);
            System.exit(1);
        } 
    }
}

JDK Flight Recorder (JFR)は、仮想スレッドに関連する次のイベントを出力できます:

• jdk.VirtualThreadStartおよびjdk.VirtualThreadEndは、仮想スレッドが開始したときと終了したときを示します。これらのイベントはデフォルトでは無効です。
• jdk.VirtualThreadPinnedは、しきい値期間より長く仮想スレッドが固定された(そしてそのキャリア・スレッドが解放されなかった)ことを示します。このイベントは、20ミリ秒のしきい値でデフォルトで有効になっています。
• jdk.VirtualThreadSubmitFailedは、おそらくリソースの問題が原因で、仮想スレッドの開始またはパーキング解除が失敗したことを示します。仮想スレッドをパーキングすると、基礎となるキャリア・スレッドが他の作業をできるように解放されます。仮想スレッドをパーキング解除すると、仮想スレッドの続行がスケジュールされます。このイベントはデフォルトで有効です。

jdk.VirtualThreadStartイベントおよびjdk.VirtualThreadEndイベントを有効にするには、JDK Mission ControlまたはカスタムJFR構成を使用します。『Java Platform, Standard Edition Flight Recorder APIプログラマーズ・ガイド』のFlight Recorderの構成を参照してください。

これらのイベントを出力するには、次のコマンドを実行します(recording.jfrは記録するためのファイル名です):

jfr print --events jdk.VirtualThreadStart,jdk.VirtualThreadEnd,jdk.VirtualThreadPinned,jdk.VirtualThreadSubmitFailed recording.jfr

プレーン・テキストとJSON形式のスレッド・ダンプを作成できます:

jcmd <PID> Thread.dump_to_file -format=text <file>
jcmd <PID> Thread.dump_to_file -format=json <file>

JSON形式は、この形式を受け入れるデバッグ・ツールにとって最適です。

jcmdスレッド・ダンプには、ネットワークI/O操作でブロックされている仮想スレッドと、ExecutorServiceインタフェースによって作成された仮想スレッドがリスト表示されます。これには、オブジェクト・アドレス、ロック、JNI統計、ヒープ統計、および従来のスレッド・ダンプに表示されるその他の情報は含まれません。

仮想スレッドは、リクエストごとのスレッド・スタイルで記述されたサーバーの(レイテンシではなく)スループットを大幅に改善できます。このスタイルでは、サーバーはスレッドをその期間全体にわたって各受信リクエストの処理専用にします。単一のリクエストを処理するときに、複数のスレッドを使用して複数のタスクを同時に実行する必要があるため、少なくとも1つのスレッドを専用にします。

プラットフォーム・スレッドをブロックすると、意味のある作業を実行していない間もスレッド(比較的少ないリソース)が占有されるため、コストが高くなります。仮想スレッドは豊富にあるため、ブロックは低コストであり推奨されます。そのため、簡単な同期スタイルでコードを記述し、ブロックI/O APIを使用する必要があります。

たとえば、非ブロッキングの非同期スタイルで記述された次のコードは、仮想スレッドの恩恵をあまり受けません。

CompletableFuture.supplyAsync(info::getUrl, pool)
   .thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofString()))
   .thenApply(info::findImage)
   .thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofByteArray()))
   .thenApply(info::setImageData)
   .thenAccept(this::process)
   .exceptionally(t -> { t.printStackTrace(); return null; });

一方、同期スタイルで記述され、単純なブロックIOを使用する次のコードは、非常に恩恵を受けます。

try {
   String page = getBody(info.getUrl(), HttpResponse.BodyHandlers.ofString());
   String imageUrl = info.findImage(page);
   byte[] data = getBody(imageUrl, HttpResponse.BodyHandlers.ofByteArray());   
   info.setImageData(data);
   process(info);
} catch (Exception ex) {
   t.printStackTrace();
}

また、このようなコードの方が、デバッガでデバッグしたり、プロファイラでプロファイルを作成したり、スレッド・ダンプで監視したりするのが簡単です。仮想スレッドを監視するには、jcmdコマンドを使用してスレッド・ダンプを作成します。

jcmd <pid> Thread.dump_to_file -format=json <file>

このスタイルで記述されるスタックが多いほど、パフォーマンスと可観測性の両方に仮想スレッドがより適しています。タスクごとにスレッドを専用にしない他のスタイルで記述されたプログラムまたはフレームワークは、仮想スレッドから大きな恩恵を受けることを期待できません。同期的なブロック・コードを非同期フレームワークと混在させないでください。

仮想スレッドについて内在化するのが最も難しいのは、プラットフォーム・スレッドと動作は同じでも、同じプログラム概念を表さないことです。

プラットフォーム・スレッドは希少であるため、貴重なリソースです。貴重なリソースは管理する必要があり、プラットフォーム・スレッドを管理する最も一般的な方法はスレッド・プールを使用することです。次に答える必要がある質問は、プール内のスレッド数です。

ただし、仮想スレッドは豊富にあるため、それぞれが、共有のプールされたリソースではなくタスクを表す必要があります。管理対象リソース・スレッドから、アプリケーション・ドメイン・オブジェクトに変わります。一連のユーザー名をメモリーに格納するために使用する文字列の数に関する質問の答えが明らかであるのと同様に、必要な仮想スレッドの数に関する質問の答えも明らかです。仮想スレッドの数は、常にアプリケーション内の同時タスクの数と等しくなります。

n個のプラットフォーム・スレッドをn個の仮想スレッドに変換しても、メリットはほとんどありません。むしろそれは変換する必要があるタスクです。

すべてのアプリケーション・タスクをスレッドとして表現するために、次の例のような共有スレッド・プール・エグゼキュータは使用しないでください。

Future<ResultA> f1 = sharedThreadPoolExecutor.submit(task1);
Future<ResultB> f2 = sharedThreadPoolExecutor.submit(task2);
// ... use futures

かわりに、次の例のような仮想スレッド・エグゼキュータを使用します。

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
   Future<ResultA> f1 = executor.submit(task1);
   Future<ResultB> f2 = executor.submit(task2);
   // ... use futures
}

コードでは引き続きExecutorServiceを使用しますが、Executors.newVirtualThreadPerTaskExecutor()から返されたコードはスレッド・プールを使用しません。かわりに、発行したタスクごとに新しい仮想スレッドが作成されます。

さらに、ExecutorService自体は軽量で、単純なオブジェクトの場合と同様に新しいものを作成できます。これにより、新しく追加されたExecutorService.close()メソッドおよびtry-with-resources構成に依存できます。tryブロックの最後に暗黙的に呼び出されるcloseメソッドは、ExecutorServiceに送信されたすべてのタスク(つまり、ExecutorServiceによって生成されたすべての仮想スレッド)が終了するまで自動的に待機します。

これは、次の例のように異なるサービスに対して複数の送信コールを同時に実行するファンアウト・シナリオで特に有用なパターンです。

void handle(Request request, Response response) {
    var url1 = ...
    var url2 = ...
 
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var future1 = executor.submit(() -> fetchURL(url1));
        var future2 = executor.submit(() -> fetchURL(url2));
        response.send(future1.get() + future2.get());
    } catch (ExecutionException | InterruptedException e) {
        response.fail(e);
    }
}
 
String fetchURL(URL url) throws IOException {
    try (var in = url.openStream()) {
        return new String(in.readAllBytes(), StandardCharsets.UTF_8);
    }
}

短期間の小さな同時タスクでも、前述のように新しい仮想スレッドを作成する必要があります。

さらに、ファンアウト・パターンおよびその他の一般的な並行処理パターンを記述し、可観測性を向上させるには、構造化並行性を使用します。

原則として、アプリケーションに10,000個以上の仮想スレッドがない場合、仮想スレッドのメリットはほとんどありません。負荷が軽すぎてスループットを向上させることができないか、または仮想スレッドに十分に多くのタスクが示されません。

特定の操作の並行処理を制限する必要がある場合があります。たとえば、外部サービスによっては、10を超える同時リクエストを処理できない場合があります。プラットフォーム・スレッドは、通常はプールで管理される貴重なリソースであるため、スレッド・プールは非常に広く存在するようになり、次の例のように、並行処理を制限するこの目的で使用されるようになりました。

ExecutorService es = Executors.newFixedThreadPool(10);
...
Result foo() {
    try {
        var fut = es.submit(() -> callLimitedService());
        return f.get();
    } catch (...) { ... }
}

この例では、限定サービスに対する同時リクエストが最大10個あることを確認します。

ただし、並行処理の制限は、スレッド・プールの操作の副次的な効果にすぎません。プールは希少なリソースを共有するように設計されており、仮想スレッドは希少ではないため、プールすべきではありません。

仮想スレッドを使用する場合、一部のサービスへのアクセスの並行処理を制限するには、その目的専用に設計された構成(Semaphoreクラス)を使用する必要があります。このクラスの例を次に示します。

Semaphore sem = new Semaphore(10);
...
Result foo() {
    sem.acquire();
    try {
        return callLimitedService();
    } finally {
        sem.release();
    }
}

fooを呼び出すために発生するスレッドは、一度に10個しか進行できないように抑制(ブロック)され、他のスレッドは、その責務から解放されます。

セマフォを使用して一部の仮想スレッドを単純にブロックすることは、タスクを固定スレッド・プールに送信することと大きく異なるように見えますが、そうではありません。スレッド・プールにタスクを送信すると、後で実行するためにキューに入れられますが、セマフォ内部(またはその問題に対する他のブロッキング同期構成)では、プールされたスレッドによって実行されるのを待機しているタスクのキューをミラー化する、ブロックされているスレッドのキューが作成されます。仮想スレッドはタスクであるため、結果の構造は同じです。

プラットフォーム・スレッドのプールはキューから取り出したタスクを処理するワーカーと考えることができ、仮想スレッドのプールは続行するまでブロックされるタスク自体と考えることができますが、コンピュータ内の基礎となる表現は事実上同じです。キューに入れられたタスクとブロックされたスレッドが同等であることを認識すると、仮想スレッドを最大限に活用できます。

データベース接続プール自体がセマフォとして機能します。10接続に制限された接続プールは、接続の取得を試行する11番目のスレッドをブロックします。接続プールの上にセマフォを追加する必要はありません。

仮想スレッドは、プラットフォーム・スレッドと同様にスレッド・ローカル変数をサポートします。詳細は、「スレッド・ローカル変数」を参照してください。通常、スレッド・ローカル変数は、現在実行中のコードに現在のトランザクションやユーザーIDなどのコンテキスト固有の情報を関連付けるために使用されます。このスレッド・ローカル変数の使用は、仮想スレッドでは完全に妥当です。ただし、より安全で効率のよいスコープ値の使用を検討してください。詳細は、「スコープ値」を参照してください。

仮想スレッドとは基本的に相容れないスレッド・ローカル変数のもう1つの用途があります。それは、再利用可能なオブジェクトのキャッシュです。通常、これらのオブジェクトは作成にコストがかかり(また大量のメモリーを消費し)、変更可能ではなく、スレッドセーフでもありません。これらは、インスタンス化される回数とメモリー内のインスタンス数を減らすためにスレッド・ローカル変数にキャッシュされますが、異なる時刻にスレッドで実行される複数のタスクによって再利用されます。

たとえば、SimpleDateFormatのインスタンスは作成にコストがかかり、スレッドセーフではありません。出現したパターンは、このようなインスタンスを次の例のようにThreadLocalにキャッシュすることです。

static final ThreadLocal<SimpleDateFormat> cachedFormatter = 
       ThreadLocal.withInitial(SimpleDateFormat::new);

void foo() {
  ...
	cachedFormatter.get().format(...);
	...
}

この種類のキャッシュが役立つのは、プラットフォーム・スレッドがプールされる場合のように、スレッド(したがって、スレッドにローカルにキャッシュされた高コストのオブジェクト)が複数のタスクによって共有および再利用される場合のみです。多くのタスクでは、スレッド・プールでの実行時にfooを呼び出すことができますが、プールには少数のスレッドしか含まれていないため、オブジェクトは数回(プール・スレッド・キャッシュごとに1回)のみインスタンス化され、キャッシュされて再利用されます。

ただし、仮想スレッドはプールされることがなく、関連のないタスクによって再利用されることはありません。すべてのタスクには独自の仮想スレッドがあるため、異なるタスクからfooを呼び出すたびに、新しいSimpleDateFormatのインスタンス化がトリガーされます。さらに、多数の仮想スレッドが同時に実行されている可能性があるため、高コストのオブジェクトはかなりのメモリーを消費する可能性があります。これらの結果は、スレッド・ローカルのキャッシュが実現しようとしているものとは正反対です。

提供する一般的な代替案は1つではありませんが、SimpleDateFormatの場合はDateTimeFormatterに置き換える必要があります。DateTimeFormatterは不変であるため、単一のインスタンスをすべてのスレッドで共有できます。

static final DateTimeFormatter formatter = DateTimeFormatter….;

void foo() {
  ...
	formatter.format(...);
	...
}

スレッド・ローカル変数を使用した高コストの共有オブジェクトのキャッシュは、非常に少数のプールされたスレッドによって使用されるという暗黙的な仮定の下で、非同期フレームワークによってバックグラウンドで実行される場合があります。これは、仮想スレッドと非同期フレームワークを混在させることがよい考えではない理由の1つです。メソッドを呼び出すと、キャッシュおよび共有されることを意図していたコストの高いオブジェクトがスレッド・ローカル変数内でインスタンス化される可能性があります。

仮想スレッドの実装の現在の制限事項は、synchronizedブロックまたはメソッドの内部でブロック操作を実行すると、JDKの仮想スレッド・スケジューラによって貴重なOSスレッドがブロックされるのに対し、ブロック操作がsynchronizedブロックまたはメソッドの外部で実行された場合はブロックされないことです。その状況を「固定」と呼びます。ブロック操作が長期間かつ頻繁な場合、固定はサーバーのスループットに悪影響を及ぼすことがあります。インメモリー操作などの短期間の操作や、synchronizedブロックまたはメソッドを使用した低頻度の操作を保護しても、悪影響はありません。

悪影響を及ぼす可能性のある固定のインスタンスを検出するために、JDK Flight Recorder (JFR)は、ブロック操作が固定されるとjdk.VirtualThreadPinnedスレッドを発行します。デフォルトでは、このイベントは、操作にかかる時間が20ミリ秒を超える場合に有効になります。

または、固定中にスレッドがブロックされたときに、システム・プロパティjdk.tracePinnedThreadsを使用してスタック・トレースを発行できます。オプション-Djdk.tracePinnedThreads=fullを指定して実行すると、スレッドが固定中にブロックされたときに完全なスタック・トレースが出力され、ネイティブ・フレームとフレーム保持モニターが強調表示されます。オプション-Djdk.tracePinnedThreads=shortを指定して実行すると、出力は問題のあるフレームだけに制限されます。

これらのメカニズムによって固定が長期間かつ頻繁な場所が検出された場合は、それらの場所でsynchronizedの使用をReentrantLockに置き換えます(繰り返しますが、短期間または低頻度の操作を保護するsynchronizedを置き換える必要はありません)。次に、syncrhonizedブロックの長期間かつ頻繁な使用の例を示します。

synchronized(lockObj) {
    frequentIO();
}

次の行に置き換えることができます。

lock.lock();
try {
    frequentIO();
} finally {
    lock.unlock();
}