CompletableFuture

Giriş

Bir proje yöneticisi düşünün. Eski usul (Future): birine iş veriyorsunuz, sonra başında dikilip bitmesini bekliyorsunuz. Yeni usul (CompletableFuture): birine iş veriyorsunuz, "bitince şunu yap, sonra şunu yap, hata olursa şöyle çöz" diye talimat bırakıyorsunuz ve kendiniz başka işlere geçiyorsunuz. İşler otomatik olarak zincirleme ilerliyor — tıpkı bir montaj hattı gibi.

Java 8 ile gelen CompletableFuture, asenkron programlamanın en güçlü aracıdır. Eski Future arayüzünün ciddi eksikliklerini — zincir oluşturamama, callback ekleyememe, birden fazla future'ı birleştirememe — gidermek için tasarlanmıştır. Spring Boot'ta @Async ile birlikte veya bağımsız olarak kullanılabilir.

Bu derste eski Future'ın sınırlarını, CompletableFuture'ın temel oluşturma yöntemlerini, dönüşüm ve zincirleme metotlarını (thenApply, thenCompose), birleştirme operasyonlarını (allOf, anyOf), hata yönetimi stratejilerini, timeout mekanizmalarını ve Spring Boot entegrasyonunu derinlemesine öğreneceğiz.

Future vs CompletableFuture

Eski Future'ın Sınırları

ExecutorService executor = Executors.newFixedThreadPool(4);

// Eski Future — son derece sınırlı
Future<String> future = executor.submit(() -> fetchDataFromApi());

// Sorun 1: Sadece bloke ederek sonuç alabilirsiniz
String result = future.get();  // Thread BLOKE olur, bekler!

// Sorun 2: Timeout ile bile bloklayıcı
String result = future.get(5, TimeUnit.SECONDS);

// Sorun 3: Zincir oluşturamazsınız
// "Veriyi al, sonra dönüştür, sonra kaydet" — tek satırda yapılamaz

// Sorun 4: İki Future'ı birleştiremezsiniz
// "Her iki sonuç da hazır olunca şunu yap" — imkansız

// Sorun 5: Hata olduğunda varsayılan değer döndüremezsiniz
// try-catch ile sarmalayıp get() çağırmanız gerekir

// Sorun 6: Manuel tamamlayamazsınız
// Bir Future'ı dışarıdan "şu sonuçla tamamla" diyemezsiniz

CompletableFuture Farkı

// CompletableFuture — Tüm eksiklikleri giderir
CompletableFuture.supplyAsync(() -> fetchUser(userId))       // 1. Asenkron başlat
    .thenApply(user -> user.getEmail())                       // 2. Dönüştür
    .thenCompose(email -> fetchOrdersByEmail(email))           // 3. Asenkron zincirle
    .thenAccept(orders -> log.info("Sipariş: {}", orders))    // 4. Tüket
    .exceptionally(ex -> {                                     // 5. Hata yönet
        log.error("Hata oluştu", ex);
        return null;
    });
// Hiçbir yerde bloke yok! Tüm zincir asenkron çalışır.

Asenkron İşlem Başlatma

supplyAsync — Değer Döndüren

// Değer döndüren asenkron işlem
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // Bu kod ayrı thread'de çalışır
    return fetchDataFromApi();
});

// Özel executor ile (önerilen — production'da mutlaka kullanın)
ExecutorService myExecutor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(
    () -> fetchDataFromApi(),
    myExecutor
);

runAsync — Değer Döndürmeyen

// Değer döndürmeyen asenkron işlem (fire-and-forget)
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    sendNotification();
});

completedFuture — Hazır Değer

// Zaten hazır olan bir değeri Future'a sarma (test veya cache senaryoları)
CompletableFuture<String> ready = CompletableFuture.completedFuture("cached-value");

⚠️ Dikkat: Executor belirtmezseniz ForkJoinPool.commonPool() kullanılır. Bu pool, tüm uygulama tarafından paylaşılır ve thread sayısı CPU çekirdek sayısı - 1 ile sınırlıdır. Bir görev bu pool'u bloke ederse tüm uygulama yavaşlar. Production'da mutlaka kendi executor'ınızı tanımlayın.

Dönüşüm Metotları

thenApply — Sonucu Dönüştür (map)

Bir CompletableFuture'ın sonucunu senkron olarak dönüştürür. Java Stream'deki map() gibi düşünün:

CompletableFuture<String> nameFuture = CompletableFuture.supplyAsync(() -> "merhaba")
    .thenApply(s -> s.toUpperCase())       // "MERHABA"
    .thenApply(s -> "Sonuç: " + s);        // "Sonuç: MERHABA"

// Zincirleme dönüşüm — her adım bir öncekinin sonucunu alır
CompletableFuture<Integer> result = CompletableFuture.supplyAsync(() -> "42")
    .thenApply(Integer::parseInt)           // "42" → 42
    .thenApply(n -> n * 2)                   // 42 → 84
    .thenApply(n -> n + 10);                 // 84 → 94

thenAccept — Sonucu Tüket (void)

Sonucu alır ama yeni bir değer döndürmez. Loglama, veritabanına kaydetme gibi yan etkiler için:

CompletableFuture.supplyAsync(() -> fetchUser(userId))
    .thenAccept(user -> {
        log.info("Kullanıcı bulundu: {}", user.getName());
        auditService.logAccess(user.getId());
    });

thenRun — Sonuçla İlgilenme

Önceki sonuçla ilgilenmez, sadece tamamlandığında bir şey yapar:

CompletableFuture.supplyAsync(() -> heavyComputation())
    .thenRun(() -> log.info("Hesaplama tamamlandı!"));

Async Varyantları

Her dönüşüm metodunun Async varyantı vardır. Fark: thenApply callback'i önceki adımla aynı thread'de çalıştırırken, thenApplyAsync farklı bir thread'de çalıştırır:

// thenApply — aynı thread'de çalışır
future.thenApply(data -> transform(data));

// thenApplyAsync — farklı thread'de çalışır (ForkJoinPool veya özel executor)
future.thenApplyAsync(data -> transform(data));
future.thenApplyAsync(data -> transform(data), myExecutor);

Ne zaman Async varyantı kullanmalı? Callback'in kendisi ağır bir işlem yapıyorsa (hesaplama, I/O) Async kullanın. Hafif dönüşümler (String manipulation, mapping) için normal varyant yeterlidir.

Zincirleme: thenCompose vs thenApply

Bu iki metot arasındaki fark, CompletableFuture'ın en kafa karıştırıcı noktasıdır. Java Stream analojisi ile açıklayalım:

• thenApply = map() — senkron dönüşüm

• thenCompose = flatMap() — asenkron dönüşüm (iç içe Future'ı düzleştirir)

// Senaryo: userId al → kullanıcıyı getir (asenkron)

// ❌ thenApply ile — İÇ İÇE FUTURE!
CompletableFuture<CompletableFuture<User>> nested =
    getUserId().thenApply(id -> fetchUser(id));  // fetchUser, CompletableFuture döndürür
// Sonuç tipi: CompletableFuture<CompletableFuture<User>> — kullanışsız!

// ✅ thenCompose ile — DÜZ FUTURE
CompletableFuture<User> flat =
    getUserId().thenCompose(id -> fetchUser(id));  // flatMap gibi düzleştirir
// Sonuç tipi: CompletableFuture<User> — temiz!

Karar Kuralı

Dönüşüm fonksiyonunuz ne döndürüyor?
  → Normal değer (String, User, int) → thenApply
  → CompletableFuture<T> → thenCompose

Gerçek Dünya Örneği — Sipariş Akışı

CompletableFuture<OrderConfirmation> orderFlow =
    validateOrder(order)                          // CompletableFuture<ValidatedOrder>
    .thenCompose(valid -> processPayment(valid))  // async → CompletableFuture<PaymentResult>
    .thenCompose(payment -> reserveStock(payment))// async → CompletableFuture<StockResult>
    .thenApply(stock -> createConfirmation(stock)) // sync dönüşüm → OrderConfirmation
    .thenApplyAsync(conf -> enrichWithTracking(conf)); // async dönüşüm (ağır işlem)

Birleştirme: allOf ve anyOf

allOf — Tüm Future'lar Tamamlandığında

Birden fazla bağımsız asenkron işlemi paralel çalıştırıp hepsinin tamamlanmasını beklemek için:

CompletableFuture<String> userFuture = fetchUserAsync(userId);          // 200ms
CompletableFuture<List<Order>> ordersFuture = fetchOrdersAsync(userId); // 150ms
CompletableFuture<Integer> balanceFuture = fetchBalanceAsync(userId);   // 100ms

// Üç istek paralel çalışır — toplam süre: max(200, 150, 100) = 200ms
CompletableFuture<Void> allDone = CompletableFuture.allOf(
    userFuture, ordersFuture, balanceFuture
);

// allOf tamamlandığında tüm sonuçlar hazırdır
allDone.thenRun(() -> {
    String user = userFuture.join();       // Artık bloke etmez (zaten bitti)
    List<Order> orders = ordersFuture.join();
    int balance = balanceFuture.join();
    buildDashboard(user, orders, balance);
});

Dikkat: allOf, CompletableFuture<Void> döndürür — bireysel sonuçları taşımaz. Sonuçlara erişmek için orijinal future'lardan .join() ile alırsınız.

Tip-Güvenli allOf Alternatifi

// allOf'un sonuç taşımaması rahatsız ediyorsa, kendi utility'nizi yazın
public static <T1, T2, T3> CompletableFuture<Triple<T1, T2, T3>> allOf(
        CompletableFuture<T1> f1,
        CompletableFuture<T2> f2,
        CompletableFuture<T3> f3) {
    return CompletableFuture.allOf(f1, f2, f3)
        .thenApply(v -> new Triple<>(f1.join(), f2.join(), f3.join()));
}

anyOf — İlk Tamamlanan (Yarış)

Birden fazla kaynaktan aynı veriyi istersiniz, ilk gelen kazanır:

// Race pattern — en hızlı kaynağı kullan
CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    fetchFromCacheAsync(),        // çok hızlı olabilir (1ms)
    fetchFromDatabaseAsync(),     // orta hızda (50ms)
    fetchFromRemoteApiAsync()     // yavaş olabilir (200ms)
);

fastest.thenAccept(result -> log.info("İlk gelen: {}", result));

⚠️ Dikkat: anyOf, CompletableFuture<Object> döndürür. Tip güvenliği için cast gerekir. Ayrıca diğer future'lar iptal edilmez — çalışmaya devam eder (kaynakları tüketir).

thenCombine — İki Future'ı Birleştir

İki bağımsız future'ın sonuçlarını birleştirmek için thenCombine kullanışlıdır:

CompletableFuture<User> userFuture = fetchUserAsync(userId);
CompletableFuture<List<Order>> ordersFuture = fetchOrdersAsync(userId);

CompletableFuture<UserDashboard> dashboard = userFuture.thenCombine(
    ordersFuture,
    (user, orders) -> new UserDashboard(user, orders)
);

Exception Handling

CompletableFuture'da hata yönetimi için üç temel metot vardır:

exceptionally — Hatada Varsayılan Değer

CompletableFuture<String> result = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) throw new RuntimeException("Sunucu hatası");
    return "başarılı";
}).exceptionally(ex -> {
    log.error("Hata: {}", ex.getMessage());
    return "varsayılan-değer";  // Hata yerine bu değer kullanılır
});

handle — Her İki Durumu da Ele Al

CompletableFuture<String> result = CompletableFuture.supplyAsync(() -> riskyOperation())
    .handle((value, ex) -> {
        if (ex != null) {
            log.error("Hata: {}", ex.getMessage());
            return "fallback";
        }
        return "Sonuç: " + value;
    });

whenComplete — Sonucu Değiştirmeden Yan Etki

CompletableFuture<String> result = CompletableFuture.supplyAsync(() -> riskyOperation())
    .whenComplete((value, ex) -> {
        if (ex != null) {
            log.error("Hata oluştu", ex);
            metricsService.recordError(ex);
        } else {
            log.info("Başarılı: {}", value);
            metricsService.recordSuccess();
        }
    });
// Sonuç DEĞİŞMEZ — sadece yan etki (loglama, metrik)

Karşılaştırma

Zincirdeki Hata Yayılımı

CompletableFuture.supplyAsync(() -> "başla")
    .thenApply(s -> {
        throw new RuntimeException("Adım 1 patladı!");
        // Bu noktadan sonraki thenApply'lar ATLANIR
    })
    .thenApply(s -> s + " — adım 2")    // ATLANIR
    .thenApply(s -> s + " — adım 3")    // ATLANIR
    .exceptionally(ex -> "Hata yakalandı: " + ex.getMessage())
    // Sonuç: "Hata yakalandı: Adım 1 patladı!"
    .thenApply(s -> s + " — devam")      // ÇALIŞIR (exceptionally hata zincirini kırar)
    // Sonuç: "Hata yakalandı: Adım 1 patladı! — devam"

Timeout (Java 9+)

// orTimeout — Süre aşılırsa TimeoutException fırlatır
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> slowOperation())
    .orTimeout(5, TimeUnit.SECONDS);
// 5 saniyede tamamlanmazsa → TimeoutException

// completeOnTimeout — Süre aşılırsa varsayılan değer döndürür
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> slowOperation())
    .completeOnTimeout("varsayılan", 5, TimeUnit.SECONDS);
// 5 saniyede tamamlanmazsa → "varsayılan" sonucu kullanılır

// Gerçek dünya: birden fazla servise timeout ile
CompletableFuture<UserProfile> profile = fetchProfileAsync(userId)
    .completeOnTimeout(UserProfile.defaultProfile(), 2, TimeUnit.SECONDS);

CompletableFuture<List<Order>> orders = fetchOrdersAsync(userId)
    .completeOnTimeout(List.of(), 2, TimeUnit.SECONDS);

Spring Boot Entegrasyonu

@Async ile Birlikte

@Service
public class DashboardService {
    
    @Async("dashboardExecutor")
    public CompletableFuture<UserProfile> fetchProfile(Long userId) {
        UserProfile profile = userClient.getProfile(userId);
        return CompletableFuture.completedFuture(profile);
    }
    
    @Async("dashboardExecutor")
    public CompletableFuture<List<Order>> fetchOrders(Long userId) {
        List<Order> orders = orderClient.getOrders(userId);
        return CompletableFuture.completedFuture(orders);
    }
    
    @Async("dashboardExecutor")
    public CompletableFuture<WalletInfo> fetchWallet(Long userId) {
        WalletInfo wallet = walletClient.getWallet(userId);
        return CompletableFuture.completedFuture(wallet);
    }
}

@RestController
@RequiredArgsConstructor
public class DashboardController {
    
    private final DashboardService dashboardService;
    
    @GetMapping("/dashboard/{userId}")
    public DashboardResponse getDashboard(@PathVariable Long userId) {
        CompletableFuture<UserProfile> profileF = dashboardService.fetchProfile(userId);
        CompletableFuture<List<Order>> ordersF = dashboardService.fetchOrders(userId);
        CompletableFuture<WalletInfo> walletF = dashboardService.fetchWallet(userId);
        
        CompletableFuture.allOf(profileF, ordersF, walletF).join();
        
        return new DashboardResponse(
            profileF.join(), ordersF.join(), walletF.join()
        );
    }
}

Yaygın Hatalar

1. join() vs get()

// get() — checked exception fırlatır (try-catch zorunlu)
try {
    String result = future.get();
} catch (InterruptedException | ExecutionException e) {
    // ...
}

// join() — unchecked exception fırlatır (CompletionException)
String result = future.join(); // Daha temiz, ama hata CompletionException'a sarılı

// ✅ ÖNERİ: CompletableFuture zincirinde join(), bağımsız çağrılarda get(timeout)

2. Common Pool Tuzağı

// ❌ YANLIŞ — commonPool tüm uygulama tarafından paylaşılır
CompletableFuture.supplyAsync(() -> blockingDatabaseCall());
// Bu blocking çağrı commonPool thread'lerini tüketir
// Diğer CompletableFuture'lar thread bulamaz → uygulama yavaşlar

// ✅ DOĞRU — Kendi executor'ınızı kullanın
ExecutorService dbExecutor = Executors.newFixedThreadPool(10);
CompletableFuture.supplyAsync(() -> blockingDatabaseCall(), dbExecutor);

3. Sonucu Almayı Unutmak

// ❌ YANLIŞ — sonuç hiç okunmuyor, exception de kaybolabilir
CompletableFuture.supplyAsync(() -> importantOperation());
// Garbage collect olabilir, exception hiç loglanmaz!

// ✅ DOĞRU — en azından hata yönetimi ekleyin
CompletableFuture.supplyAsync(() -> importantOperation())
    .exceptionally(ex -> {
        log.error("Kritik hata", ex);
        return null;
    });

Gerçek Dünya Örneği: Ürün Detay Sayfası

E-ticaret ürün detay sayfası: ürün bilgisi, stok durumu, yorumlar, benzer ürünler, fiyat geçmişi — hepsi farklı servislerden gelir:

@Service
@RequiredArgsConstructor
public class ProductDetailService {
    
    private final ProductClient productClient;
    private final InventoryClient inventoryClient;
    private final ReviewClient reviewClient;
    private final RecommendationClient recommendationClient;
    private final PriceHistoryClient priceHistoryClient;
    private final Executor productExecutor;
    
    public ProductDetailResponse getProductDetail(String productId) {
        // 5 servis PARALEL çağrılır
        CompletableFuture<Product> productF = CompletableFuture
            .supplyAsync(() -> productClient.getProduct(productId), productExecutor);
        
        CompletableFuture<StockInfo> stockF = CompletableFuture
            .supplyAsync(() -> inventoryClient.getStock(productId), productExecutor)
            .completeOnTimeout(StockInfo.unknown(), 2, TimeUnit.SECONDS);
        
        CompletableFuture<List<Review>> reviewsF = CompletableFuture
            .supplyAsync(() -> reviewClient.getReviews(productId), productExecutor)
            .completeOnTimeout(List.of(), 3, TimeUnit.SECONDS);
        
        CompletableFuture<List<Product>> similarF = CompletableFuture
            .supplyAsync(() -> recommendationClient.getSimilar(productId), productExecutor)
            .completeOnTimeout(List.of(), 3, TimeUnit.SECONDS);
        
        CompletableFuture<PriceHistory> priceF = CompletableFuture
            .supplyAsync(() -> priceHistoryClient.getHistory(productId), productExecutor)
            .completeOnTimeout(PriceHistory.empty(), 2, TimeUnit.SECONDS);
        
        // Ürün bilgisi ZORUNLU, diğerleri opsiyonel (timeout ile fallback)
        Product product = productF.join(); // Bu başarısız olursa exception
        
        // Diğerleri tamamlanmasını bekle (timeout zaten ayarlandı)
        CompletableFuture.allOf(stockF, reviewsF, similarF, priceF).join();
        
        return ProductDetailResponse.builder()
            .product(product)
            .stock(stockF.join())
            .reviews(reviewsF.join())
            .similarProducts(similarF.join())
            .priceHistory(priceF.join())
            .build();
    }
}

Sonuç: Senkron: 200+150+300+250+100 = 1 saniye → Asenkron: max = 300ms → %70 hızlanma! Üstelik yavaş servisler timeout ile sınırlandırılır, sayfa her zaman gösterilir.

Özet

• CompletableFuture, Java'nın en güçlü asenkron aracıdır. Eski Future'ın tüm eksikliklerini — zincirleme, callback, birleştirme, hata yönetimi — çözer.

• supplyAsync değer döndüren, runAsync değer döndürmeyen asenkron işlemler başlatır. Production'da mutlaka kendi executor'ınızı tanımlayın.

• thenApply senkron dönüşüm (map), thenCompose asenkron zincirleme (flatMap) yapar. Fonksiyon CompletableFuture dönüyorsa thenCompose kullanın.

• allOf tüm future'ları bekler (paralel çalışma), anyOf ilk tamamlananı alır (yarış). thenCombine iki future'ı birleştirir.

• Hata yönetimi: exceptionally hatada varsayılan değer, handle her iki durumu ele alır, whenComplete yan etki (loglama) için kullanılır.

• Java 9+ orTimeout ve completeOnTimeout ile zaman aşımı yönetimi yapılır. Production'da her asenkron çağrıya timeout ekleyin.

• ForkJoinPool.commonPool() tuzağına düşmeyin — blocking I/O çağrıları common pool'u tıkar. Her zaman özel executor kullanın.