Многопоточность в Java с Project Loom: виртуальные или обычные потоки

// Создание традиционного потока
Thread thread = new Thread(() -> {
    System.out.println("Выполняется в отдельном потоке ОС");
    // Выполнение задачи
});
thread.start();

ExecutorService executor = Executors.newFixedThreadPool(200);
for (int i = 0; i < 10000; i++) {
    executor.submit(() -> {
        // Обработка запроса
    });
}

CompletableFuture.supplyAsync(() -> fetchDataFromDatabase())
    .thenApply(data -> processData(data))
    .thenAccept(result -> sendResponse(result))
    .exceptionally(ex -> handleError(ex));

// Создание виртуального потока
Thread vThread = Thread.startVirtualThread(() -> {
    System.out.println("Выполняется в виртуальном потоке");
    // Выполнение задачи
});

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    // Для каждого из тысяч запросов запускаем отдельный виртуальный поток
    for (int i = 0; i < 10000; i++) {
        executor.submit(() -> processRequest());
    }
}

// С точки зрения разработчика, следующий код выглядит блокирующим
try (var connection = dataSource.getConnection();
     var statement = connection.prepareStatement(sql)) {
    try (var resultSet = statement.executeQuery()) {
        while (resultSet.next()) {
            // Обработка результатов
        }
    }
}
// Но во время выполнения JVM освободит поток-носитель, когда запрос будет отправлен
// и вернет его, когда результаты будут готовы

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<Data> data = scope.fork(() -> fetchData());
    Future<Config> config = scope.fork(() -> fetchConfig());
    
    // Ожидаем завершения всех задач или первой ошибки
    scope.join();
    
    // Проверяем на ошибки
    scope.throwIfFailed();
    
    // Теперь можно безопасно использовать результаты
    processResults(data.resultNow(), config.resultNow());
}

// Традиционный подход с пулом потоков
ExecutorService executorService = Executors.newFixedThreadPool(200);
for (Request request : requests) {
    executorService.submit(() -> {
        Response response = sendHttpRequest(request); // Блокирующий вызов
        processResponse(response);
    });
}
 
// Подход с использованием виртуальных потоков
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (Request request : requests) {
        executor.submit(() -> {
            Response response = sendHttpRequest(request); // Всё ещё блокирующий вызов, но JVM оптимизирует его
            processResponse(response);
        });
    }
}

// Бенчмарк для традиционных потоков
long start = System.currentTimeMillis();
try (var executor = Executors.newFixedThreadPool(200)) {
    List<Future<?>> futures = new ArrayList<>();
    for (int i = 0; i < 10_000; i++) {
        futures.add(executor.submit(() -> {
            Thread.sleep(1000); // Имитация блокирующей операции
            return null;
        }));
    }
    for (Future<?> future : futures) {
        future.get();
    }
}
System.out.println("Время выполнения (традиционные потоки): " + (System.currentTimeMillis() - start) + " мс");
 
// Бенчмарк для виртуальных потоков
start = System.currentTimeMillis();
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<?>> futures = new ArrayList<>();
    for (int i = 0; i < 10_000; i++) {
        futures.add(executor.submit(() -> {
            Thread.sleep(1000); // Имитация блокирующей операции
            return null;
        }));
    }
    for (Future<?> future : futures) {
        future.get();
    }
}
System.out.println("Время выполнения (виртуальные потоки): " + (System.currentTimeMillis() - start) + " мс");

// Тест с традиционными потоками
for (int i = 0; i < 10000; i++) {
    new Thread(() -> {
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                break;
            }
        }
    }).start();
}
 
// Тот же тест с виртуальными потоками
for (int i = 0; i < 10000; i++) {
    Thread.startVirtualThread(() -> {
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                break;
            }
        }
    });
}

// Было: традиционный пул потоков
ExecutorService executor = Executors.newFixedThreadPool(200);
 
// Стало: использование виртуальных потоков
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

// Традиционный подход - ограниченный пул соединений
HikariConfig config = new HikariConfig();
config.setMaximumPoolSize(50);
DataSource dataSource = new HikariDataSource(config);
 
// С виртуальными потоками можно использовать отдельное соединение на запрос
// без значительных накладных расходов
try (Connection conn = DriverManager.getConnection(url, user, password)) {
  // Выполнение запроса
}

ExecutorService highPriorityExecutor = Executors.newThreadPerTaskExecutor(
  Thread.ofVirtual().name("high-priority-", 0).factory());
 
ExecutorService regularExecutor = Executors.newThreadPerTaskExecutor(
  Thread.ofVirtual().name("regular-", 0).factory());

ThreadFactory virtualFactory = Thread.ofVirtual()
  .name("custom-", 0)
  .uncaughtExceptionHandler((t, e) -> log.error("Unhandled in " + t.getName(), e))
  .factory();
 
int parallelism = Runtime.getRuntime().availableProcessors();
ExecutorService executor = new ForkJoinPool(parallelism,
  ForkJoinPool.defaultForkJoinWorkerThreadFactory,
  null, true) {
  @Override
  public void execute(Runnable task) {
    submit(virtualFactory.newThread(task));
  }
};

List<String> urls = List.of(/* тысячи URL-адресов */);
List<String> results = Collections.synchronizedList(new ArrayList<>());
 
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
  urls.forEach(url -> executor.submit(() -> {
    try {
      String content = fetchUrl(url);
      String extracted = extractData(content);
      results.add(extracted);
    } catch (Exception e) {
      log.error("Error processing URL: " + url, e);
    }
  }));
}
 
// Обработка собранных результатов
processResults(results);

MessageQueue queue = /* source of messages */;
MessageProcessor processor = new MessageProcessor();
 
// Обработка каждого сообщения в отдельном виртуальном потоке
while (true) {
  Message message = queue.receive();
  Thread.startVirtualThread(() -> {
    try {
      processor.process(message);
      message.acknowledge();
    } catch (Exception e) {
      message.retry();
      log.error("Processing failed", e);
    }
  });
}

// Для IO-задач
ExecutorService ioExecutor = Executors.newVirtualThreadPerTaskExecutor();
 
// Для CPU-интенсивных задач
int cpuCores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuExecutor = Executors.newFixedThreadPool(cpuCores);

// Пример интеграции Reactor с виртуальными потоками
Flux<Data> dataStream = Flux.fromIterable(sources)
  .flatMap(source -> Mono.fromCallable(() -> {
      // Этот блок может выполняться в виртуальном потоке
      return fetchDataFromSource(source);
  }).subscribeOn(Schedulers.fromExecutor(Executors.newVirtualThreadPerTaskExecutor()))
);

@Bean
public TomcatProtocolHandlerCustomizer<?> protocolHandlerVirtualThreadExecutorCustomizer() {
    return protocolHandler -> {
        protocolHandler.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
    };
}