Task yang dikirim ke ExecutorService berjalan di thread lain — pertanyaannya, bagaimana pemanggil menerima hasilnya? Jawaban pertama Java adalah interface Future: pegangan tugas yang bisa ditanyai statusnya, dibatalkan, dan diambil hasilnya. Namun Future menyimpan satu kelemahan mendasar: satu-satunya cara mengambil nilai adalah get() yang memblokir thread pemanggil. Menggabungkan hasil dua task, menjalankan tahapan lanjutan begitu hasil tersedia, atau memulihkan kegagalan tanpa memblokir — semuanya harus ditulis manual dengan get() bersarang. CompletableFuture hadir untuk mengisi kekosongan itu: sebuah Future yang bisa diprogram secara deklaratif dengan rantai tahapan. Bab ini mempraktikkan keduanya secara berurutan.

Future Klasik: Menunggu, Tenggat, Pembatalan #

Future dikembalikan oleh submit() dan invokeAll(). Tiga kemampuannya — get(), get() dengan tenggat, dan cancel() — dipraktikkan program berikut:

java FutureBlocking.java 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class FutureBlocking {
    public static void main(String[] args) throws Exception {
        ExecutorService eksekutor = Executors.newFixedThreadPool(2);
        try {
            // Task cepat: get() memblokir pemanggil sampai hasil tersedia.
            Future<Integer> penjumlahan = eksekutor.submit(() -> {
                int total = 0;
                for (int i = 1; i <= 100; i++) {
                    total += i;
                }
                return total;
            });
            System.out.println("hasil penjumlahan  : " + penjumlahan.get());

            // Task 200 ms: get dengan tenggat 1 detik -- cukup longgar.
            Future<String> pembayaran = eksekutor.submit(() -> {
                Thread.sleep(200);
                return "pembayaran terkonfirmasi";
            });
            System.out.println("status pembayaran  : "
                    + pembayaran.get(1, TimeUnit.SECONDS));

            // Task 5 detik: dibatalkan lebih awal lewat interrupt.
            Future<String> sinkronisasi = eksekutor.submit(() -> {
                Thread.sleep(5_000);
                return "sinkronisasi selesai";
            });
            Thread.sleep(50); // pastikan task sudah mulai berjalan
            boolean dibatalkan = sinkronisasi.cancel(true);
            System.out.println("sinkronisasi dibatalkan: " + dibatalkan);
            System.out.println("status future: dibatalkan=" + sinkronisasi.isCancelled()
                    + ", selesai=" + sinkronisasi.isDone());
        } finally {
            eksekutor.shutdownNow(); // tidak ada task lagi yang perlu ditunggu
        }
    }
}

Output yang Dihasilkan:

text 
hasil penjumlahan  : 5050
status pembayaran  : pembayaran terkonfirmasi
sinkronisasi dibatalkan: true
status future: dibatalkan=true, selesai=true

Mari kita bedah tiga pola pengambilan hasil tersebut:

  1. penjumlahan.get(): thread main berhenti di baris ini sampai task selesai, lalu menerima 5050. Selama menunggu, thread main tidak bisa mengerjakan apa pun — blokade total.
  2. pembayaran.get(1, TimeUnit.SECONDS): bentuk bertenggat. Bila task tidak selesai dalam 1 detik, TimeoutException terlempar dan pemanggil bisa mengambil keputusan; di sini task hanya 200 milidetik sehingga hasil pembayaran diterima sebagaimana mestinya.
  3. sinkronisasi.cancel(true): argumen true berarti task yang sedang berjalan di-interupsi — Thread.sleep(5_000) di dalamnya terbangun lewat InterruptedException. Setelah dibatalkan, isCancelled() dan isDone() sama-sama true: task yang dibatalkan dianggap selesai (dengan kegagalan), dan get() padanya akan melempar CancellationException.

CompletableFuture: Rantai Tanpa Blokade #

CompletableFuture mengubah cara berpikir dari “menunggu hasil lalu mengolah” menjadi “menempelkan tahapan pengolahan pada hasil yang akan datang”. Method supplyAsync() memulai rantai di sebuah executor; thenApply() memetakan nilai ke bentuk baru; thenCombine() menyatukan dua rantai; exceptionally() menyiapkan nilai cadangan saat tahapan mana pun gagal:

java CompletableFutureRantai.java 
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureRantai {
    public static void main(String[] args) {
        // Executor kustom: tahapan berat tidak menumpuk di common pool.
        ExecutorService eksekutor = Executors.newFixedThreadPool(3);
        try {
            // Rantai 1: ambil total item, gandakan, format menjadi teks.
            // thenApply menjalankan tahap berikutnya TANPA memblokir pemanggil.
            CompletableFuture<String> laporan = CompletableFuture
                    .supplyAsync(() -> 250, eksekutor)
                    .thenApply(total -> total * 2)
                    .thenApply(jumlah -> "total: " + jumlah + " item");

            // Rantai 2: berjalan paralel dengan rantai 1.
            CompletableFuture<String> pelanggan = CompletableFuture
                    .supplyAsync(() -> "Siti", eksekutor)
                    .thenApply(nama -> "pelanggan: " + nama);

            // Kombinasi hasil dua rantai menjadi satu keluaran.
            CompletableFuture<String> gabungan = laporan.thenCombine(
                    pelanggan, (bagianLaporan, bagianPelanggan) ->
                            bagianPelanggan + ", " + bagianLaporan);

            // Rantai dengan pemulihan galat: sumber diskon sengaja melempar
            // exception; exceptionally menyediakan nilai cadangan.
            CompletableFuture<Integer> diskon = CompletableFuture
                    .<Integer>supplyAsync(() -> {
                        throw new IllegalStateException("promosi tidak aktif");
                    }, eksekutor)
                    .exceptionally(e -> 0);

            // Tunggu seluruh rantai tuntas, lalu cetak hasil terurut.
            CompletableFuture.allOf(gabungan, diskon).join();
            System.out.println(gabungan.join());
            System.out.println("diskon terpakai: " + diskon.join() + "%");
        } finally {
            eksekutor.shutdown();
        }
    }
}

Output yang Dihasilkan:

text 
pelanggan: Siti, total: 500 item
diskon terpakai: 0%

Bedah baris-per-baris untuk rantai tersebut:

  1. supplyAsync(() -> 250, eksekutor): tahap pertama dijalankan di executor kustom. Tanpa argumen kedua, tahap berjalan di ForkJoinPool.commonPool() — pool bersama seluruh JVM yang dibahas khusus pada Chapter Parallel Streams dan Concurrency.
  2. Dua thenApply berurutan: pemetaan 250 → 500 → "total: 500 item". Tahap berikutnya dijalankan oleh thread yang menyelesaikan tahap sebelumnya — pemanggil tidak pernah menunggu; main terus berjalan sampai titik join().
  3. thenCombine: menunggu kedua rantai (yang berjalan paralel sejak awal) lalu menggabungkan dua hasilnya dengan fungsi (a, b) -> b + ", " + a. Hasilnya pelanggan: Siti, total: 500 item.
  4. exceptionally(e -> 0): IllegalStateException dari tahap supplyAsync tidak pernah terlihat oleh pemanggil; rantai dilanjutkan dengan nilai cadangan 0. Tanpa baris ini, join() pada rantai diskon melempar CompletionException.
  5. allOf(...).join() lalu cetak terurut: allOf mengembalikan future yang selesai ketika seluruh anggotanya selesai — gerbang tunggu tunggal sebelum pencetakan. Setelah gerbang itu, tiap join() langsung kembali dengan nilai yang sudah tersimpan, dan urutan cetak ditentukan oleh urutan pemanggilan.

join versus get #

Kedua method sama-sama menunggu hasil, tetapi berbeda pada jenis exception. get() adalah method warisan Future: melempar tiga checked exception (InterruptedException, ExecutionException, TimeoutException) sehingga pemanggil wajib menangkap semuanya. join() milik CompletableFuture melempar CompletionException — exception runtime yang membungkus penyebab aslinya — sehingga rantai dan stream tetap bersih tanpa blok try bersarang. Konvensi praktis: pakai join() di dalam pipeline, pakai get(timeout) di tepi sistem yang membutuhkan tenggat.

Keluarga Tahapan dalam Sekilas #

MethodPeran
supplyAsync / runAsyncMemulai rantai dengan hasil / aksi
thenApply / thenAccept / thenRunTransformasi nilai / konsumsi nilai / aksi lanjutan
thenCombineMenyatukan hasil dua rantai
thenComposeMerangkai tahap yang hasilnya juga CompletableFuture (flat-map)
allOf / anyOfMenunggu semua / salah satu rantai selesai
exceptionally / whenCompleteNilai cadangan / catatan samping saat kegagalan
complete / completeExceptionallyMengisi hasil dari luar (future sebagai janji)

Rantai asinkron menyusun alur kerja; namun data yang dipakai task-task tersebut sering kali berupa Map atau List bersama yang diakses banyak thread sekaligus. Pustaka java.util.concurrent menyiapkan struktur data khusus untuk itu — topik Chapter berikutnya.

Referensi #

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

CompletableFutureRantai.java

Unduh berkas
java 
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureRantai {
    public static void main(String[] args) {
        // Executor kustom: tahapan berat tidak menumpuk di common pool.
        ExecutorService eksekutor = Executors.newFixedThreadPool(3);
        try {
            // Rantai 1: ambil total item, gandakan, format menjadi teks.
            // thenApply menjalankan tahap berikutnya TANPA memblokir pemanggil.
            CompletableFuture<String> laporan = CompletableFuture
                    .supplyAsync(() -> 250, eksekutor)
                    .thenApply(total -> total * 2)
                    .thenApply(jumlah -> "total: " + jumlah + " item");

            // Rantai 2: berjalan paralel dengan rantai 1.
            CompletableFuture<String> pelanggan = CompletableFuture
                    .supplyAsync(() -> "Siti", eksekutor)
                    .thenApply(nama -> "pelanggan: " + nama);

            // Kombinasi hasil dua rantai menjadi satu keluaran.
            CompletableFuture<String> gabungan = laporan.thenCombine(
                    pelanggan, (bagianLaporan, bagianPelanggan) ->
                            bagianPelanggan + ", " + bagianLaporan);

            // Rantai dengan pemulihan galat: sumber diskon sengaja melempar
            // exception; exceptionally menyediakan nilai cadangan.
            CompletableFuture<Integer> diskon = CompletableFuture
                    .<Integer>supplyAsync(() -> {
                        throw new IllegalStateException("promosi tidak aktif");
                    }, eksekutor)
                    .exceptionally(e -> 0);

            // Tunggu seluruh rantai tuntas, lalu cetak hasil terurut.
            CompletableFuture.allOf(gabungan, diskon).join();
            System.out.println(gabungan.join());
            System.out.println("diskon terpakai: " + diskon.join() + "%");
        } finally {
            eksekutor.shutdown();
        }
    }
}

FutureBlocking.java

Unduh berkas
java 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class FutureBlocking {
    public static void main(String[] args) throws Exception {
        ExecutorService eksekutor = Executors.newFixedThreadPool(2);
        try {
            // Task cepat: get() memblokir pemanggil sampai hasil tersedia.
            Future<Integer> penjumlahan = eksekutor.submit(() -> {
                int total = 0;
                for (int i = 1; i <= 100; i++) {
                    total += i;
                }
                return total;
            });
            System.out.println("hasil penjumlahan  : " + penjumlahan.get());

            // Task 200 ms: get dengan tenggat 1 detik -- cukup longgar.
            Future<String> pembayaran = eksekutor.submit(() -> {
                Thread.sleep(200);
                return "pembayaran terkonfirmasi";
            });
            System.out.println("status pembayaran  : "
                    + pembayaran.get(1, TimeUnit.SECONDS));

            // Task 5 detik: dibatalkan lebih awal lewat interrupt.
            Future<String> sinkronisasi = eksekutor.submit(() -> {
                Thread.sleep(5_000);
                return "sinkronisasi selesai";
            });
            Thread.sleep(50); // pastikan task sudah mulai berjalan
            boolean dibatalkan = sinkronisasi.cancel(true);
            System.out.println("sinkronisasi dibatalkan: " + dibatalkan);
            System.out.println("status future: dibatalkan=" + sinkronisasi.isCancelled()
                    + ", selesai=" + sinkronisasi.isDone());
        } finally {
            eksekutor.shutdownNow(); // tidak ada task lagi yang perlu ditunggu
        }
    }
}