thenCompose + власний Executor + тайм-аути

1. thenCompose vs. thenApply: різниця і коли що використовувати

В асинхронному програмуванні на Java (через CompletableFuture) часто потрібно виконувати ланцюжки дій. Для цього є два схожі методи — thenApply і thenCompose. Але працюють вони по‑різному!

thenApply

Метод thenApply використовується, коли наступний крок — це просте перетворення значення, без запуску нових асинхронних операцій. Він отримує результат попереднього кроку, обробляє його та повертає нове значення (а не CompletableFuture).

Якщо ви знайомі зі Stream API, то thenApply поводиться приблизно як map: бере результат, застосовує функцію та повертає перетворений варіант.

Приклад:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "42");
CompletableFuture<Integer> lengthFuture = cf.thenApply(s -> s.length());
// lengthFuture містить 2 (довжина рядка "42")

Простіше кажучи, thenApply — це спосіб сказати: «Коли результат буде готовий, зроби з ним ось це».

thenCompose

• Використовується, коли наступний крок — ще одна асинхронна операція (повертає CompletableFuture).
• Дозволяє «розгортати» вкладені CompletableFuture (аналог flatMap).
• Якщо використати thenApply з асинхронною функцією, вийде CompletableFuture<CompletableFuture<T>> — незручно!

Приклад:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "user42");

// Припустимо, нам потрібно за іменем користувача отримати його замовлення (асинхронно)
CompletableFuture<List<Order>> ordersFuture = cf.thenCompose(username -> fetchOrdersAsync(username));
// fetchOrdersAsync повертає CompletableFuture<List<Order>>

Візуально:

• thenApply: CF<String> → thenApply(s -> s.length()) → CF<Integer>
• thenCompose: CF<User> → thenCompose(u -> fetchOrdersAsync(u.id)) → CF<List<Order>>

Коли що використовувати?

• Функція повертає звичайне значення — використовуйте thenApply.
• Функція повертає CompletableFuture — використовуйте thenCompose.

Приклад помилки:

cf.thenApply(username -> fetchOrdersAsync(username)); // Отримаєте CF<CF<List<Order>>>
cf.thenCompose(username -> fetchOrdersAsync(username)); // Отримаєте CF<List<Order>>

2. Керування пулом потоків (Executor): навіщо і як використовувати власний Executor

За замовчуванням: ForkJoinPool.commonPool()

Коли ви пишете CompletableFuture.supplyAsync(...) або thenApplyAsync(...) без зазначення Executor, Java використовує спільний пул потоків — ForkJoinPool.commonPool(). Це зручно, але підходить не завжди:

• Якщо у вас багато тривалих або блокувальних операцій (мережеві запити, робота з файлами), спільний пул може перевантажитися, і всі задачі чекатимуть.
• Іноді потрібно ізолювати задачі з різними пріоритетами або обмежити кількість одночасно працюючих потоків.

Коли потрібен власний Executor?

• Тривалі, блокувальні операції (наприклад, запити до БД, HTTP‑запити, читання файлів).
• Ізоляція задач: щоб користувацькі задачі не заважали системним.
• Обмеження ресурсів: наприклад, не запускати понад 10 одночасних завантажень.

Як створити власний Executor

Зазвичай використовують ThreadPoolExecutor або фабрики з Executors:

ExecutorService myExecutor = Executors.newFixedThreadPool(10);

Як використовувати власний Executor із CompletableFuture

• У методах supplyAsync, runAsync, thenApplyAsync, thenComposeAsync та інших можна передати другим аргументом ваш Executor.

Приклади:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(
    () -> loadDataFromNetwork(), myExecutor
);

cf.thenApplyAsync(data -> processData(data), myExecutor)
  .thenAcceptAsync(result -> System.out.println(result), myExecutor);

Важливо: якщо не вказати Executor, буде використано ForkJoinPool.commonPool().

Коли достатньо типового Executor?

• Для коротких, CPU-bound задач (прості обчислення).
• Коли неважливо, у якому потоці виконується задача.

3. Обробка тайм-аутів: orTimeout і completeOnTimeout

Асинхронні операції можуть зависнути або виконуватися надто довго (наприклад, якщо сервер не відповідає). Щоб не чекати безкінечно, у CompletableFuture є методи для роботи з тайм-аутами.

orTimeout

• Завершує CompletableFuture з винятком TimeoutException, якщо операція не завершилась за вказаний час.
• Не скасовує фактично виконувану задачу, але downstream‑ланцюжок отримає помилку.

Синтаксис:

cf.orTimeout(3, TimeUnit.SECONDS)
  .exceptionally(ex -> {
      System.out.println("Тайм-аут: " + ex);
      return null;
  });

Приклад:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(5000); // імітуємо тривалу операцію
    return "OK";
});

cf.orTimeout(2, TimeUnit.SECONDS)
  .exceptionally(ex -> {
      System.out.println("Помилка: " + ex);
      return "TIMEOUT";
  });

Результат:

Через 2 секунди буде згенеровано TimeoutException, і exceptionally обробить помилку.

completeOnTimeout

• Завершує CompletableFuture із зазначеним значенням, якщо операція не завершилась за час тайм-ауту.
• Не кидає виняток, а повертає «резервне» значення.

Синтаксис:

cf.completeOnTimeout("DEFAULT", 2, TimeUnit.SECONDS);

Приклад:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(5000);
    return "OK";
});

cf.completeOnTimeout("TIMEOUT", 2, TimeUnit.SECONDS)
  .thenAccept(System.out::println); // Через 2 секунди виведе "TIMEOUT"

Порівняння orTimeout і completeOnTimeout

orTimeout

completeOnTimeout

4. Практика: приклад із thenCompose, власним Executor і тайм-аутом

Задача:

• Отримати користувача за ID (асинхронно, із затримкою).
• Потім асинхронно отримати список замовлень користувача (також із затримкою).
• Використати власний Executor.
• Додати тайм-аут на отримання замовлень.

import java.util.concurrent.*;
import java.util.*;

public class AsyncDemo {
    static ExecutorService ioExecutor = Executors.newFixedThreadPool(4);

    // Імітація асинхронного отримання користувача
    static CompletableFuture<String> fetchUserAsync(int userId) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            return "user" + userId;
        }, ioExecutor);
    }

    // Імітація асинхронного отримання замовлень користувача
    static CompletableFuture<List<String>> fetchOrdersAsync(String username) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(3000); // Тривала операція!
            return List.of("order1", "order2");
        }, ioExecutor);
    }

    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException ignored) {}
    }

    public static void main(String[] args) {
        fetchUserAsync(42)
            .thenCompose(username ->
                fetchOrdersAsync(username)
                    .orTimeout(2, TimeUnit.SECONDS) // Тайм-аут на отримання замовлень
                    .exceptionally(ex -> {
                        System.out.println("Не вдалося отримати замовлення: " + ex);
                        return List.of();
                    })
            )
            .thenAccept(orders -> System.out.println("Замовлення: " + orders))
            .join(); // Чекаємо завершення всього ланцюжка

        ioExecutor.shutdown();
    }
}

Що відбувається:

• Отримуємо користувача (1 секунда).
• Отримуємо замовлення (3 секунди, але тайм-аут 2 секунди).
• Якщо не встигли — перехоплюємо TimeoutException, повертаємо порожній список.
• Усе працює на власному Executor.

Результат:

Не вдалося отримати замовлення: java.util.concurrent.TimeoutException
Замовлення: []

Якщо зменшити затримку в fetchOrdersAsync до 1_000 мс — побачите реальні замовлення.

5. Типові помилки й нюанси

Помилка № 1: Використання thenApply замість thenCompose для асинхронних операцій.
Якщо функція повертає CompletableFuture, а ви застосували thenApply, отримаєте вкладений тип CompletableFuture<CompletableFuture<T>>. Це ускладнить ланцюжок і призведе до зайвих обгорток. Рішення: використовуйте thenCompose, щоб «сплющити» результат у CompletableFuture<T>.

Помилка № 2: Запуск тривалих або IO‑задач без власного Executor.
За замовчуванням задачі виконуються в ForkJoinPool.commonPool(). Якщо його перевантажити, затримки почнуть зростати, а інші задачі в застосунку можуть сповільнитися. Рішення: створюйте власний ExecutorService і передавайте його в supplyAsync/thenApplyAsync.

Помилка № 3: Очікування, що orTimeout скасовує виконання задачі.
orTimeout лише завершує CompletableFuture з винятком за тайм-аутом, але сама задача продовжує працювати у фоні. Рішення: якщо потрібно зупинити виконання, використовуйте cancel(true) або власні механізми переривання.

Помилка № 4: Неправильне розуміння області дії тайм-ауту.
orTimeout і completeOnTimeout працюють лише для одного конкретного кроку ланцюжка, а не для всього ланцюга. Рішення: якщо потрібен загальний тайм-аут на весь ланцюжок, обгорніть його в окремий CompletableFuture і застосуйте тайм-аут до нього.

Помилка № 5: Не закрито ExecutorService.
Якщо після виконання задач не викликати shutdown()/shutdownNow() у ExecutorService, потоки продовжать працювати, і програма може «зависнути». Рішення: завжди закривайте ExecutorService у finally або використовуйте try-with-resources у Java 21+.

1

Опитування

Асинхронне програмування, рівень 55, лекція 4

Недоступний

Асинхронне програмування

Асинхронне програмування

Відкрити