Using the ExecutorService for virtual threads – Concurrency – Virtual Threads, Structured Concurrency

212. Using the ExecutorService for virtual threads

Virtual threads allow us to write more expressive and straightforward concurrent code. Thanks to the massive throughput obtained via virtual threads we can easily adopt the task-per-thread model (for an HTTP server, this means a request per thread, for a database, this means a transaction per thread, and so on). In other words, we can assign a new virtual thread for each concurrent task. Trying to use the task-per-thread model with platform threads will result in a throughput limited by the number of hardware’s cores – this is explained by Little’s law (https://en.wikipedia.org/wiki/Little%27s_law), L = λW, or throughput equals average concurrency divided by latency.Whenever possible it is recommended to avoid interacting with threads directly. JDK sustains this via ExecutorService/Executor API. More precisely, we are used to submitting a task (Runnable/Callable) to an ExecutorService/Executor and working with the returned Future. This pattern is valid for virtual threads as well.So, we don’t have to write ourselves all the plumbing code for adopting the task-per-thread for virtual threads because, starting with JDK 19, this model is available via the Executors class. More precisely, via the newVirtualThreadPerTaskExecutor() method which creates an ExecutorService capable to create an unbounded number of virtual threads that follows the task-per-thread model. This ExecutorService exposes methods that allow us to give the tasks such as the submit() (as you’ll see next) and invokeAll/Any() (as you’ll see later) methods, and return a Future containing an exception or a result.

Starting with JDK 19, the ExecutorService extends the AutoCloseable interface. In other words, we can use ExecutorService in a try-with-resources pattern.

Consider the following simple Runnable and Callable:

Runnable taskr = () -> logger.info(
  Thread.currentThread().toString());
      
Callable<Boolean> taskc = () -> {
  logger.info(Thread.currentThread().toString());
  return true;
};

Executing the Runnable/Callable can be done as follows (here, we submit 15 tasks (NUMBER_OF_TASKS = 15)):

try (ExecutorService executor
      = Executors.newVirtualThreadPerTaskExecutor()) {
  for (int i = 0; i < NUMBER_OF_TASKS; i++) {
    executor.submit(taskr); // executing Runnable
    executor.submit(taskc); // executing Callable
  }
}

Of course, in the case of Runnable/Callable we can capture a Future and act accordingly via the blocking get() method or whatever we want to do.

Future<?> future = executor.submit(taskr);
Future<Boolean> future = executor.submit(taskc);

A possible output looks as follows:

VirtualThread[#28]/runnable@ForkJoinPool-1-worker-6
VirtualThread[#31]/runnable@ForkJoinPool-1-worker-5
VirtualThread[#29]/runnable@ForkJoinPool-1-worker-7
VirtualThread[#25]/runnable@ForkJoinPool-1-worker-3
VirtualThread[#24]/runnable@ForkJoinPool-1-worker-2
VirtualThread[#27]/runnable@ForkJoinPool-1-worker-5
VirtualThread[#26]/runnable@ForkJoinPool-1-worker-4
VirtualThread[#22]/runnable@ForkJoinPool-1-worker-1
VirtualThread[#36]/runnable@ForkJoinPool-1-worker-1
VirtualThread[#37]/runnable@ForkJoinPool-1-worker-2
VirtualThread[#35]/runnable@ForkJoinPool-1-worker-7
VirtualThread[#34]/runnable@ForkJoinPool-1-worker-4
VirtualThread[#32]/runnable@ForkJoinPool-1-worker-3
VirtualThread[#33]/runnable@ForkJoinPool-1-worker-2
VirtualThread[#30]/runnable@ForkJoinPool-1-worker-1

Check out the virtual threads’ ids. They range between #22-#37 without repetition. Each task is executed by its own virtual thread.The task-per-thread model is also available for classical threads via newThreadPerTaskExecutor(ThreadFactory threadFactory). Here is an example:

static class SimpleThreadFactory implements ThreadFactory {
  @Override
  public Thread newThread(Runnable r) {
    return new Thread(r);                     // classic
   // return Thread.ofVirtual().unstarted(r); // virtual
  }
}
try (ExecutorService executor =
      Executors.newThreadPerTaskExecutor(
          new SimpleThreadFactory())) {
  for (int i = 0; i < NUMBER_OF_TASKS; i++) {
    executor.submit(taskr); // executing Runnable
    executor.submit(taskc); // executing Callable
  }
}

As you can see, newThreadPerTaskExecutor() can be used for classic threads or for virtual threads. The number of created threads is unbounded. By simply modifying the thread factory we can switch between virtual/classic threads.A possible output looks as follows:

Thread[#75,Thread-15,5,main]
Thread[#77,Thread-17,5,main]
Thread[#76,Thread-16,5,main]
Thread[#83,Thread-23,5,main]
Thread[#82,Thread-22,5,main]
Thread[#80,Thread-20,5,main]
Thread[#81,Thread-21,5,main]
Thread[#79,Thread-19,5,main]
Thread[#78,Thread-18,5,main]
Thread[#89,Thread-29,5,main]
Thread[#88,Thread-28,5,main]
Thread[#87,Thread-27,5,main]
Thread[#86,Thread-26,5,main]
Thread[#85,Thread-25,5,main]
Thread[#84,Thread-24,5,main]

Check out the threads’ ids. They range between #75-#89 without repetition. Each task is executed by its own thread.