JavaのVirtual Threads入門 - 軽量スレッドで並行処理を簡単に

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// プラットフォームスレッド（固定プール）の場合
// 200スレッドのプールでは、10,000タスク完了に約50秒
try (var executor = Executors.newFixedThreadPool(200)) {
    IntStream.range(0, 10_000).forEach(i -> {
        executor.submit(() -> {
            Thread.sleep(Duration.ofSeconds(1));
            return i;
        });
    });
}

// Virtual Threadsの場合
// 10,000スレッドが並行動作し、約1秒で完了
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    IntStream.range(0, 10_000).forEach(i -> {
        executor.submit(() -> {
            Thread.sleep(Duration.ofSeconds(1));
            return i;
        });
    });
}

1
2
3
4
5
6
7
8
9

// 仮想スレッドを作成して即座に開始
Thread thread = Thread.startVirtualThread(() -> {
    System.out.println("Hello from Virtual Thread!");
    System.out.println("Thread name: " + Thread.currentThread().getName());
    System.out.println("Is virtual: " + Thread.currentThread().isVirtual());
});

// スレッドの終了を待機
thread.join();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// Thread.Builderを使用してVirtual Threadを作成
Thread.Builder builder = Thread.ofVirtual().name("my-virtual-thread");

Runnable task = () -> {
    System.out.println("Running in: " + Thread.currentThread().getName());
};

// unstartedで未開始のスレッドを作成
Thread thread = builder.unstarted(task);
System.out.println("Thread created but not started");

// 明示的に開始
thread.start();
thread.join();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// "worker-0", "worker-1", ... という名前のスレッドを作成
Thread.Builder builder = Thread.ofVirtual().name("worker-", 0);

Runnable task = () -> {
    System.out.println("Thread ID: " + Thread.currentThread().threadId() +
                       ", Name: " + Thread.currentThread().getName());
};

// worker-0
Thread t1 = builder.start(task);
t1.join();

// worker-1
Thread t2 = builder.start(task);
t2.join();

// worker-2
Thread t3 = builder.start(task);
t3.join();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// Virtual Threads用のThreadFactoryを作成
ThreadFactory factory = Thread.ofVirtual()
    .name("request-handler-", 0)
    .factory();

// ThreadFactoryからスレッドを生成
Thread t1 = factory.newThread(() -> {
    System.out.println("Handling request in: " + Thread.currentThread().getName());
});
t1.start();
t1.join();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    // 複数のタスクを投入
    Future<String> future1 = executor.submit(() -> {
        Thread.sleep(Duration.ofMillis(500));
        return "Result 1";
    });
    
    Future<String> future2 = executor.submit(() -> {
        Thread.sleep(Duration.ofMillis(300));
        return "Result 2";
    });
    
    Future<String> future3 = executor.submit(() -> {
        Thread.sleep(Duration.ofMillis(100));
        return "Result 3";
    });
    
    // 結果を取得
    System.out.println(future1.get());
    System.out.println(future2.get());
    System.out.println(future3.get());
}
// try-with-resourcesでclose()が呼ばれ、すべてのタスク完了を待機

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

void handle(Request request, Response response) {
    var url1 = URI.create("https://api.example.com/users").toURL();
    var url2 = URI.create("https://api.example.com/orders").toURL();
    
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var userFuture = executor.submit(() -> fetchURL(url1));
        var orderFuture = executor.submit(() -> fetchURL(url2));
        
        // 両方の結果を待機して結合
        String result = userFuture.get() + orderFuture.get();
        response.send(result);
    } 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);
    }
}

1
2
3
4
5
6
7

// 悪い例：Virtual Threadsをプールする
// 意味がないだけでなく、パフォーマンスに悪影響
ExecutorService badExecutor = Executors.newFixedThreadPool(100, 
    Thread.ofVirtual().factory());

// 良い例：タスクごとに新しいVirtual Threadを作成
ExecutorService goodExecutor = Executors.newVirtualThreadPerTaskExecutor();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// ピンニングが発生する例（非推奨）
synchronized (lock) {
    // この中でブロッキング操作を行うと、キャリアスレッドがブロックされる
    Thread.sleep(Duration.ofSeconds(1));  // ピンニング発生
}

// ピンニングを回避する例（推奨）
private final ReentrantLock lock = new ReentrantLock();

lock.lock();
try {
    // ReentrantLockを使用すれば、ブロッキング操作でもアンマウント可能
    Thread.sleep(Duration.ofSeconds(1));  // ピンニング回避
} finally {
    lock.unlock();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

public class RateLimitedService {
    // 同時実行数を10に制限
    private final Semaphore semaphore = new Semaphore(10);
    
    public String callExternalService(String request) throws InterruptedException {
        semaphore.acquire();  // 許可を取得（ブロック可能）
        try {
            return doCallExternalService(request);
        } finally {
            semaphore.release();  // 許可を解放
        }
    }
    
    private String doCallExternalService(String request) {
        // 実際の外部サービス呼び出し
        return "Response for: " + request;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

// プレビュー機能を有効にする必要がある
// java --enable-preview --source 21 Main.java

import java.util.concurrent.StructuredTaskScope;

Response handle() throws ExecutionException, InterruptedException {
    // ShutdownOnFailureは、いずれかのサブタスクが失敗したら全体をシャットダウン
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        // サブタスクをフォーク（新しいVirtual Threadで実行）
        Supplier<String> user = scope.fork(() -> findUser());
        Supplier<Integer> order = scope.fork(() -> fetchOrder());
        
        // すべてのサブタスクの完了を待機
        scope.join();
        
        // エラーがあれば例外をスロー
        scope.throwIfFailed();
        
        // 両方成功した場合、結果を取得
        return new Response(user.get(), order.get());
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// ShutdownOnSuccess: 最初に成功したサブタスクの結果を使用
<T> T race(List<Callable<T>> tasks, Instant deadline) 
        throws InterruptedException, ExecutionException, TimeoutException {
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<T>()) {
        for (var task : tasks) {
            scope.fork(task);
        }
        // デッドラインまで待機
        return scope.joinUntil(deadline).result();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

// 移行前：固定スレッドプール
public class OldHttpClient {
    private final ExecutorService executor = Executors.newFixedThreadPool(100);
    
    public List<String> fetchAll(List<URL> urls) throws Exception {
        List<Future<String>> futures = new ArrayList<>();
        for (URL url : urls) {
            futures.add(executor.submit(() -> fetch(url)));
        }
        
        List<String> results = new ArrayList<>();
        for (Future<String> future : futures) {
            results.add(future.get());
        }
        return results;
    }
}

// 移行後：Virtual Threads
public class NewHttpClient {
    public List<String> fetchAll(List<URL> urls) throws Exception {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = urls.stream()
                .map(url -> executor.submit(() -> fetch(url)))
                .toList();
            
            List<String> results = new ArrayList<>();
            for (Future<String> future : futures) {
                results.add(future.get());
            }
            return results;
        }
    }
}

1
2

# JSONフォーマットでスレッドダンプを出力
jcmd <pid> Thread.dump_to_file -format=json thread-dump.json