Java 21 LTS Release + Virtual Threads = Death of Thread Pools?

Java 21 was the 4th Long Term Support (LTS) release of Java after Java 8 and was released on 19th of September 2023.

This release removed the preview status of Virtual Threads and now made it an official part of the JDK. If you want to learn more about Java Virtual Threads you can read my previous post.

In short Java Virtual Threads are very inexpensive to create and destroy as they are not associated one to one Platform threads but instead many to one Platform threads.

This makes Virtual Thread basically a part of fire and forget idiom as opposed to create a fixed amount and reuse (pooling) idiom. Not only that Java has officially requested users of Virtual Threads to not to pool.

But there will be times when you need some sort of mechanism to atleast mimic the pooling capability of Thread pool. For example in a microservice architecture a call to a downstream service must be limited to avoid latency and throughput overhead.

In this type of scenarios a Thread pool would be ideal as it will not allow more than a fixed size of concurrent threads calling the downstream service.

Since we can not/must not use Thread pools with Virtual threads. We can write a class using decorator pattern which encapsulates a Virtual Thread executor by makes it behave as a Fixed size thread pool.

UPDATE: An update was done after the comment from Pavlo Ivanitskyi

public class FixedVirtualThreadExecutorService implements ExecutorService {
    private final ExecutorService VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE = Executors.newVirtualThreadPerTaskExecutor();

    private Semaphore semaphore;

    private int poolSize;

    public FixedVirtualThreadExecutorService(int poolSize) {
        this.poolSize = poolSize;
        this.semaphore = new Semaphore(this.poolSize);
    }

    @Override
    public void shutdown() {
        VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        return VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.shutdownNow();
    }

    @Override
    public boolean isShutdown() {
        return VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.isShutdown();
    }

    @Override
    public boolean isTerminated() {
        return VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.isTerminated();
    }

    @Override
    public boolean awaitTermination(long timeout, java.util.concurrent.TimeUnit unit) throws InterruptedException {
        return VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.awaitTermination(timeout, unit);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                return task.call();
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        }, VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE);

    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                task.run();
                return result;
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        }, VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE);
    }

    @Override
    public Future<?> submit(Runnable task) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                task.run();
                return null;
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        }, VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE);
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        return tasks.stream().map(t -> CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                return t.call();
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        }, VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE)).collect(Collectors.toList());
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, java.util.concurrent.TimeUnit unit) throws InterruptedException {
        return tasks.stream().map(t -> CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                return t.call();
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        }, VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE).orTimeout(timeout, unit)).collect(Collectors.toList());
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        return tasks.stream().map(t -> CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                return t.call();
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        }, VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE)).map(f -> {
            try {
                return f.get();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }).findAny().get();
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, java.util.concurrent.TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return invokeAll(tasks, timeout, unit).stream().map(f -> {
            try {
                return f.get();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }).findAny().get();
    }

    @Override
    public void close() {
        VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.close();
    }

    @Override
    public void execute(Runnable command) {
        VIRTUAL_THREAD_POOL_EXECUTOR_SERVICE.execute(() -> {
            try {
                semaphore.acquire();
                command.run();
            } catch (Exception e) {
                throw new IllegalStateException(e);
            } finally {
                semaphore.release();
            }
        });

    }
}

In this case we are making use of a Semaphore to control access to the underlying Virtual Thread executor. Now you can make use of this and make call to underlying service without overwhelming it.

ExecutorService executorService = new FixedVirtualThreadExecutorService(10);
final RestTemplate restTemplate = new RestTemplate();
for (int i=0;i<20;i++) {
    executorService.execute(() -> {
        System.out.println(restTemplate.getForObject("https://www.google.com?q=Shazin", String.class));
    });
}

In this case even if we call the execute method 20 times, it will make sure to restrict concurrent calls to maximum of 10.