Java 21 仮想スレッドの実践活用 — 従来スレッドとの違いと移行指針

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;

public class VirtualThreadsDemo {

    /** 外部 API 呼び出しを模擬（I/O バウンド） */
    static String callApi(String endpoint) {
        try {
            Thread.sleep(100); // ネットワーク遅延を模擬
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "Response from " + endpoint
                + " [" + Thread.currentThread() + "]";
    }

    /** --- 1. 基本: Thread.ofVirtual() で仮想スレッドを生成 --- */
    static void basicVirtualThread() throws InterruptedException {
        var thread = Thread.ofVirtual()
                .name("my-virtual-thread")
                .start(() -> {
                    System.out.println("仮想スレッド: "
                            + Thread.currentThread());
                    System.out.println("isVirtual: "
                            + Thread.currentThread().isVirtual());
                });
        thread.join();
    }

    /** --- 2. Thread.startVirtualThread() で簡易起動 --- */
    static void quickStart() throws InterruptedException {
        var thread = Thread.startVirtualThread(() ->
                System.out.println("簡易起動: "
                        + Thread.currentThread().isVirtual()));
        thread.join();
    }

    /** --- 3. ExecutorService で大量タスクを並行実行 --- */
    static void massiveConcurrency() throws Exception {
        var taskCount = 1000;
        var start = Instant.now();

        try (var executor = Executors
                .newVirtualThreadPerTaskExecutor()) {
            var futures = new ArrayList<Future<String>>();
            for (var i = 0; i < taskCount; i++) {
                var endpoint = "/api/resource/" + i;
                futures.add(executor.submit(
                        () -> callApi(endpoint)));
            }

            var successCount = 0;
            for (var future : futures) {
                future.get();
                successCount++;
            }
            var elapsed = Duration.between(start, Instant.now());
            System.out.println(successCount + " タスク完了: "
                    + elapsed.toMillis() + " ms");
        }
    }

    /** --- 4. pinning 回避: synchronized → ReentrantLock --- */
    static class SafeCounter {
        private final ReentrantLock lock = new ReentrantLock();
        private int count = 0;

        /** ReentrantLock なら仮想スレッドが pinning されない */
        void increment() {
            lock.lock();
            try {
                count++;
            } finally {
                lock.unlock();
            }
        }

        int getCount() {
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }
    }

    /** --- 5. プラットフォームスレッドとの性能比較 --- */
    static long benchmark(ExecutorService executor, int tasks)
            throws Exception {
        var start = Instant.now();
        var futures = new ArrayList<Future<String>>();
        for (var i = 0; i < tasks; i++) {
            var endpoint = "/api/bench/" + i;
            futures.add(executor.submit(
                    () -> callApi(endpoint)));
        }
        for (var f : futures) {
            f.get();
        }
        return Duration.between(start, Instant.now()).toMillis();
    }

    public static void main(String[] args) throws Exception {
        System.out.println("=== 1. 基本的な仮想スレッド ===");
        basicVirtualThread();

        System.out.println("\n=== 2. 簡易起動 ===");
        quickStart();

        System.out.println("\n=== 3. 1000タスク並行実行 ===");
        massiveConcurrency();

        System.out.println("\n=== 4. pinning 回避カウンター ===");
        var counter = new SafeCounter();
        try (var executor = Executors
                .newVirtualThreadPerTaskExecutor()) {
            var futures = new ArrayList<Future<?>>();
            for (var i = 0; i < 10000; i++) {
                futures.add(executor.submit(counter::increment));
            }
            for (var f : futures) {
                f.get();
            }
        }
        System.out.println("カウンター結果: " + counter.getCount()
                + " (期待値: 10000)");

        System.out.println("\n=== 5. 性能比較 ===");
        var tasks = 500;

        try (var platformPool = Executors
                .newFixedThreadPool(50)) {
            var platformMs = benchmark(platformPool, tasks);
            System.out.println("プラットフォームスレッド (50本): "
                    + platformMs + " ms");
        }

        try (var virtualPool = Executors
                .newVirtualThreadPerTaskExecutor()) {
            var virtualMs = benchmark(virtualPool, tasks);
            System.out.println("仮想スレッド: "
                    + virtualMs + " ms");
        }
    }
}