Kenapa penting

Di bab sebelumnya kamu membuat Thread sendiri dan menjaga data bersama dengan lock. Di aplikasi nyata, hampir tidak ada yang menulis new Thread(...) di kode bisnis. Membuat thread itu mahal, jumlahnya harus dibatasi, dan error di dalamnya mudah hilang. Yang dipakai sehari-hari adalah executor (pengelola thread) dan CompletableFuture (hasil yang akan tersedia nanti). Pola paling umum di backend: satu endpoint perlu data dari beberapa layanan — produk, harga, stok — yang masing-masing butuh ratusan milidetik. Memanggilnya berurutan berarti waktunya dijumlahkan; memanggilnya paralel berarti waktunya kira-kira sama dengan layanan paling lambat. Bab ini mengajarkan cara melakukannya dengan aman, termasuk menutup executor, menangani error, dan memberi batas waktu.

Konsep inti

Thread pool dan ExecutorService

Thread pool adalah sekumpulan thread yang dibuat sekali lalu dipakai ulang. Kamu tidak lagi berurusan dengan thread, melainkan dengan tugas (Runnable atau Callable<T>) yang dikirim ke ExecutorService. Pabriknya ada di class Executors:

Pabrik	Perilaku
Executors.newFixedThreadPool(n)	Tepat n thread; tugas berlebih mengantre. Cocok untuk kerja CPU-bound.
Executors.newVirtualThreadPerTaskExecutor()	Satu virtual thread baru per tugas, tanpa batas pool. Cocok untuk I/O.
Executors.newSingleThreadExecutor()	Satu thread; tugas dijalankan berurutan.

Virtual thread dibahas mendalam di bab berikutnya. Untuk sekarang cukup tahu: executor ini murah dan ideal untuk tugas yang banyak menunggu (HTTP, database).

submit, Future, dan invokeAll

submit(callable) langsung kembali dan memberi Future<T> — tiket untuk mengambil hasil nanti. future.get() memblokir sampai hasil siap; bila tugas melempar exception, get() melemparnya kembali terbungkus ExecutionException. invokeAll(daftarTugas) mengirim banyak tugas sekaligus dan menunggu semuanya selesai.

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

public class ExecutorDemo {
    static int slowSquare(int n) throws InterruptedException {
        Thread.sleep(200);
        return n * n;
    }

    public static void main(String[] args) throws Exception {
        long start = System.nanoTime();
        try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
            Future<Integer> f = pool.submit(() -> slowSquare(7));
            System.out.println("submit selesai, isDone=" + f.isDone());
            System.out.println("hasil=" + f.get());

            List<Callable<Integer>> tasks = List.of(
                () -> slowSquare(1), () -> slowSquare(2), () -> slowSquare(3));
            int total = 0;
            for (Future<Integer> r : pool.invokeAll(tasks)) {
                total += r.get();
            }
            System.out.println("total=" + total);
        } // close(): tunggu semua tugas selesai, lalu shutdown
        System.out.println("waktu ~" + (System.nanoTime() - start) / 100_000_000 * 100 + " ms");
    }
}

submit selesai, isDone=false
hasil=49
total=14
waktu ~400 ms

Satu tugas 200 ms, lalu tiga tugas paralel 200 ms — totalnya ±400 ms, bukan 800 ms.

Menutup executor dengan benar

Executor yang tidak di-shutdown menyisakan thread hidup: memori bocor, dan pada thread non-daemon JVM bahkan tidak mau berhenti. Sejak Java 19, ExecutorService meng-extend AutoCloseable, sehingga cara termudah adalah try-with-resources seperti contoh di atas. close() memanggil shutdown() (menolak tugas baru) lalu menunggu tugas yang sudah dikirim selesai.

Di kode lama kamu akan melihat pola manual shutdown() → awaitTermination(timeout) → shutdownNow() bila melewati batas waktu. Pola itu masih relevan untuk executor berumur panjang (misalnya milik sebuah komponen yang hidup selama aplikasi berjalan), sedangkan try-with-resources cocok untuk executor berumur pendek di dalam satu operasi.

TIP

Di aplikasi Spring, executor jangka panjang sebaiknya dikelola sebagai bean agar dimatikan otomatis saat aplikasi berhenti. Jangan membuat thread pool baru di setiap request.

CompletableFuture: merangkai langkah asinkron

Future biasa hanya bisa ditunggu dengan get(). CompletableFuture<T> bisa dirangkai: "kalau hasil ini sudah ada, lakukan itu". Metode yang paling sering dipakai:

Metode	Fungsi
supplyAsync(supplier, executor)	Jalankan supplier secara asinkron di executor tertentu
thenApply(fn)	Ubah hasil (seperti map pada stream)
thenCombine(other, (a, b) -> ...)	Gabungkan dua hasil independen setelah keduanya selesai
allOf(f1, f2, ...)	Future baru yang selesai saat semua future selesai
exceptionally(ex -> fallback)	Pulihkan dari error dengan nilai pengganti
orTimeout(n, unit)	Gagal dengan TimeoutException bila belum selesai dalam waktu itu
join()	Tunggu hasil; error dilempar sebagai CompletionException

allOf hanya memberi CompletableFuture<Void>. Nilainya diambil dari tiap future setelah semuanya selesai:

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

public class AllOfDemo {
    public static void main(String[] args) {
        try (ExecutorService io = Executors.newVirtualThreadPerTaskExecutor()) {
            List<String> skus = List.of("A-1", "B-2", "C-3");
            List<CompletableFuture<String>> futures = skus.stream()
                .map(sku -> CompletableFuture.supplyAsync(() -> sku + ":OK", io))
                .toList();
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            List<String> results = futures.stream().map(CompletableFuture::join).toList();
            System.out.println(results);

            CompletableFuture<Integer> failing = CompletableFuture
                .supplyAsync(() -> Integer.parseInt("abc"), io);
            try {
                failing.join();
            } catch (CompletionException e) {
                System.out.println("penyebab: " + e.getCause());
            }
        }
    }
}

[A-1:OK, B-2:OK, C-3:OK]
penyebab: java.lang.NumberFormatException: For input string: "abc"

Perhatikan bahwa error asli ada di getCause(). Saat menulis log, catat penyebabnya, bukan hanya CompletionException.

Jebakan ForkJoinPool.commonPool()

Bila kamu memanggil supplyAsync(supplier) tanpa executor, tugas dijalankan di ForkJoinPool.commonPool() — pool bersama milik seluruh JVM, yang ukurannya kira-kira jumlah core CPU dikurangi satu. Pool ini juga dipakai oleh parallel stream. Jika kamu menaruh panggilan HTTP atau query database (yang menunggu ratusan milidetik) di sana, beberapa request saja sudah cukup untuk menghabiskan semua thread-nya, dan bagian lain aplikasi yang memakai pool yang sama ikut tersendat. Aturannya sederhana: untuk I/O, selalu berikan executor sendiri sebagai argumen kedua.

Contoh di dunia backend

Halaman detail produk di toko online butuh data dari tiga layanan: katalog (±300 ms), harga (±400 ms), dan stok (sedang bermasalah, ±1.500 ms). Layanan stok tidak boleh membuat seluruh halaman menunggu, jadi diberi batas waktu 1 detik dengan fallback.

import java.math.BigDecimal;
import java.util.concurrent.*;

public class CheckoutAggregator {
    record Product(String sku, String name) {}
    record Price(BigDecimal amount) {}
    record Stock(int available) {}
    record ProductView(String name, BigDecimal price, int stock) {}

    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    // Simulasi panggilan ke tiga layanan lain
    static Product fetchProduct(String sku) { sleep(300); return new Product(sku, "Kopi Arabika 250g"); }
    static Price fetchPrice(String sku)     { sleep(400); return new Price(new BigDecimal("85000")); }
    static Stock fetchStock(String sku)     { sleep(1500); return new Stock(12); }

    public static void main(String[] args) {
        try (ExecutorService io = Executors.newVirtualThreadPerTaskExecutor()) {
            String sku = "KOPI-250";
            long start = System.nanoTime();

            CompletableFuture<Product> product =
                CompletableFuture.supplyAsync(() -> fetchProduct(sku), io);
            CompletableFuture<Price> price =
                CompletableFuture.supplyAsync(() -> fetchPrice(sku), io);
            CompletableFuture<Integer> stock =
                CompletableFuture.supplyAsync(() -> fetchStock(sku), io)
                    .thenApply(Stock::available)
                    .orTimeout(1, TimeUnit.SECONDS)
                    .exceptionally(ex -> {
                        System.out.println("stok gagal: " + ex.getClass().getSimpleName());
                        return 0; // fallback: anggap stok tidak diketahui
                    });

            ProductView view = product
                .thenCombine(price, (p, pr) -> new ProductView(p.name(), pr.amount(), 0))
                .thenCombine(stock, (v, s) -> new ProductView(v.name(), v.price(), s))
                .join();

            long ms = (System.nanoTime() - start) / 1_000_000;
            System.out.println(view);
            System.out.println("selesai dalam ~" + (ms / 100 * 100) + " ms");
        }
    }
}

stok gagal: TimeoutException
ProductView[name=Kopi Arabika 250g, price=85000, stock=0]
selesai dalam ~1000 ms

Tanpa paralelisme, halaman ini butuh 300 + 400 + 1.500 = 2.200 ms. Dengan CompletableFuture, katalog dan harga selesai bersamaan dalam ±400 ms, dan stok dipotong di 1 detik. Ada satu hal yang perlu dicatat: orTimeout hanya membuat future-nya gagal; tugas fetchStock di belakangnya tetap berjalan sampai selesai. Untuk HTTP sungguhan, pasang juga timeout di klien HTTP-nya agar koneksi benar-benar diputus.

Pertanyaan interview

1. Apa beda Future dan CompletableFuture?

Future hanya bisa dicek (isDone) dan ditunggu secara blocking (get). CompletableFuture mengimplementasikan Future sekaligus CompletionStage, sehingga bisa dirangkai (thenApply, thenCombine), digabung (allOf), diberi penanganan error (exceptionally), dan batas waktu (orTimeout) tanpa memblokir thread di setiap langkah.

2. Kenapa supplyAsync tanpa executor berbahaya untuk panggilan I/O?

Karena tugas jatuh ke ForkJoinPool.commonPool(), pool kecil yang dipakai bersama seluruh JVM (termasuk parallel stream). Panggilan blocking menahan thread-nya, sehingga pool cepat habis dan bagian lain aplikasi ikut lambat. Solusinya: berikan executor khusus I/O, misalnya newVirtualThreadPerTaskExecutor().

3. Bagaimana cara mematikan ExecutorService dengan benar?

Untuk executor berumur pendek, pakai try-with-resources (Java 19+): close() melakukan shutdown lalu menunggu semua tugas selesai. Untuk executor berumur panjang, panggil shutdown(), tunggu dengan awaitTermination sampai batas waktu tertentu, lalu shutdownNow() bila masih ada tugas yang menggantung.

Jebakan umum

• Lupa menutup executor. Thread pool yang dibuat per request lalu tidak ditutup akan menumpuk sampai aplikasi kehabisan memori atau thread.
• Memanggil get()/join() terlalu cepat. supplyAsync(a).join() lalu supplyAsync(b).join() sama saja dengan berurutan. Mulai semua tugas dulu, baru tunggu.
• Menelan exception. Error di dalam tugas tidak muncul di mana pun sampai kamu memanggil get/join atau memasang exceptionally. Tugas yang hasilnya tidak pernah dibaca bisa gagal diam-diam.
• Mengira orTimeout menghentikan pekerjaan. Ia hanya menggagalkan future; panggilan di belakangnya tetap berjalan. Pasang timeout di klien HTTP/database juga.
• I/O di commonPool. supplyAsync tanpa executor untuk panggilan blocking membuat seluruh JVM tersendat.
• Mengubah data bersama dari banyak tugas tanpa sinkronisasi. Executor tidak menghilangkan race condition dari bab sebelumnya; lebih aman setiap tugas mengembalikan hasil lalu digabung di akhir.

Ringkasan

• ExecutorService mengelola thread untukmu; kamu cukup mengirim tugas Runnable/Callable.
• submit memberi Future; invokeAll menjalankan banyak tugas dan menunggu semuanya.
• Sejak Java 19, tutup executor berumur pendek dengan try-with-resources — close() menunggu tugas selesai.
• CompletableFuture merangkai langkah asinkron: supplyAsync → thenApply/thenCombine → join, dengan allOf untuk banyak future sekaligus.
• Tangani kegagalan dengan exceptionally dan batasi waktu dengan orTimeout; error asli ada di getCause().
• Untuk I/O, selalu berikan executor sendiri — jangan biarkan jatuh ke ForkJoinPool.commonPool().

Bacaan lanjutan

• Java SE 25 — ExecutorService (Javadoc)
• Java SE 25 — CompletableFuture (Javadoc)
• Java SE 25 Core Libraries — Concurrency