Когда начинает вычисление CompletableFuture?

Когда начинается вычисление CompletableFuture

CompletableFuture — это инструмент для асинхронного программирования в Java, но моментом запуска вычислений нужно управлять явно. Это частый источник ошибок.

Ключевое правило: ничего не начнётся само по себе

// Этот код вычисления НЕ запускает!
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("This will NOT print immediately!");
    return 42;
});
// Ничего не произойдёт, пока ты не подпишешься

CompletableFuture ленив: вычисление начинается только когда:

1. Ты вызовешь блокирующий метод (.get(), .join())
2. Ты зарегистрируешь callback (.thenAccept(), .thenApply())
3. Ты объединишь его с другим future (.thenCompose(), .thenCombine())

Сценарий 1: Блокирующий вызов

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("Computing...");
    return 42;
});

// Вычисление НАЧНЁТСЯ ТОЛЬКО ЗДЕСЬ
Integer result = future.get(); // блокируем, ждём результата
System.out.println("Result: " + result);

// Output:
// Computing...
// Result: 42

Важно: .get() блокирует текущий поток! Это не асинхронность.

Сценарий 2: Регистрация callback

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("Computing...");
    return 42;
});

// Callback зарегистрирован
future.thenAccept(result -> {
    System.out.println("Result received: " + result);
});

// Вычисление начнётся асинхронно
// Ваш код не блокируется
System.out.println("Main thread continues...");

Thread.sleep(1000); // даём время на выполнение

// Output:
// Main thread continues...
// Computing...
// Result received: 42

Сценарий 3: Цепочка операций

CompletableFuture.supplyAsync(() -> {
    System.out.println("Step 1: Fetching data...");
    return 10;
})
.thenApply(value -> {
    System.out.println("Step 2: Processing " + value);
    return value * 2;
})
.thenAccept(result -> {
    System.out.println("Step 3: Result is " + result);
});

Thread.sleep(1000);

// Output:
// Step 1: Fetching data...
// Step 2: Processing 10
// Step 3: Result is 20

Все операции связаны: если предыдущая не завершена, следующая не начнётся.

Опасность: потеря future

// НЕПРАВИЛЬНО
CompletableFuture.supplyAsync(() -> {
    System.out.println("This might not run!");
    return 42;
});
// future тут теряется, вычисление может не начаться

// ПРАВИЛЬНО
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("This will run");
    return 42;
});
future.thenAccept(result -> System.out.println(result));

Thread Pool и Executor

По умолчанию supplyAsync() использует ForkJoinPool.commonPool(), но можно указать свой:

ExecutorService executor = Executors.newFixedThreadPool(2);

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("Running on thread: " + Thread.currentThread().getName());
    return 42;
}, executor); // явно указали executor

future.thenAccept(System.out::println);
Thread.sleep(100);
executor.shutdown();

Важно: если не указать executor — по умолчанию используется ForkJoinPool, который может быть bottleneck при большом количестве futures.

Синхронные vs асинхронные callbacks

ExecutorService executor = Executors.newFixedThreadPool(2);

CompletableFuture.supplyAsync(() -> {
    System.out.println("1. Computing on " + Thread.currentThread().getName());
    return 42;
})
// thenAccept выполнится на ТОМ ЖЕ потоке (синхронно)
.thenAccept(result -> {
    System.out.println("2. Processing on " + Thread.currentThread().getName());
})
// thenAcceptAsync выполнится на ДРУГОМ потоке (асинхронно)
.thenAcceptAsync(result -> {
    System.out.println("3. Final on " + Thread.currentThread().getName());
}, executor);

Thread.sleep(500);
executor.shutdown();

// Output (может быть):
// 1. Computing on ForkJoinPool.commonPool-worker-1
// 2. Processing on ForkJoinPool.commonPool-worker-1 (ТОТ ЖЕ ПОТОК!)
// 3. Final on pool-1-thread-1 (ДРУГОЙ ПОТОК)

Практический пример: фетч данных с API

public class ApiClient {
    private ExecutorService executor = Executors.newFixedThreadPool(10);
    
    public CompletableFuture<User> fetchUser(String id) {
        // Вычисление НЕ начинается здесь
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("Fetching user " + id);
            return callRestApi("/users/" + id); // имитация
        }, executor)
        .thenApply(json -> parseJson(json))
        .exceptionally(e -> {
            System.err.println("Error: " + e.getMessage());
            return new User(); // fallback
        });
    }
    
    public void example() {
        // Вычисление НАЧНЁТСЯ здесь при подписке
        fetchUser("123")
            .thenAccept(user -> System.out.println("Got user: " + user.getName()));
        
        // Или здесь при блокирующем вызове
        try {
            User user = fetchUser("456").get(5, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            System.out.println("Timeout!");
        }
    }
}

Памятка

Частая ошибка

// НЕПРАВИЛЬНО
CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> f2 = f1.thenApply(x -> x * 2);
CompletableFuture<Integer> f3 = f2.thenApply(x -> x + 5);
// Ничего не произойдёт!

// ПРАВИЛЬНО: зарегистрируй callback
f3.thenAccept(System.out::println);
// Или заблокируй
Integer result = f3.join();

Итог: CompletableFuture — это не запуск вычисления, это контейнер для результата и средство связи между асинхронными операциями. Вычисление начинается только явно, когда ты подписываешься на результат.