CompletableFuture

Future yang diperkenalkan framework executor menyelesaikan pekerjaannya dengan satu cara, yaitu get() yang memblokir pemanggilnya sampai hasilnya siap. CompletableFuture memperluas idenya, yaitu hasil yang datang nanti dapat disusun, digabung, dan diberi jalur error tanpa ada satu thread pun yang berdiri menunggu. Topik ini menyusun alur asinkron pertama dengan thenApply, menggabungkan dua alur dengan thenCombine, lalu menutup jalur errornya dengan exceptionally.

Menyusun alur dengan thenApply#

Alur asinkron dimulai dari supplyAsync, yaitu titik masuk yang menjalankan Supplier di thread lain dan membungkus hasilnya dalam CompletableFuture. Metode thenApply kemudian mendaftarkan pengubahan atas hasil itu, sehingga beberapa langkah dapat dirangkai seperti pipeline tanpa satu pun pemanggilan get() yang memblokir. Tugas-tugas yang tidak diberi executor sendiri berjalan pada ForkJoinPool.commonPool(), kumpulan thread bersama milik JVM yang juga dipakai stream paralel.

import java.util.List;
import java.util.concurrent.CompletableFuture;

public class DemoCompletableFuture {

    public static void main(String[] args) {
        CompletableFuture<Double> rataRata = CompletableFuture
                .supplyAsync(() -> List.of(80, 75, 90, 85))
                .thenApply(nilai -> nilai.stream().mapToInt(Integer::intValue).average().orElse(0));

        CompletableFuture<String> laporan = rataRata
                .thenApply(r -> "Rata-rata nilai ujian: " + r);

        System.out.println("Alur tersusun, main tidak diblokir oleh get()");

        System.out.println(rataRata.join());
        System.out.println(laporan.join());
    }
}

Alur tersusun, main tidak diblokir oleh get()
82.5
Rata-rata nilai ujian: 82.5

Kedua pernyataan pertama hanya menyusun alur, yaitu menghitung rata-rata lalu memformatnya, dan tidak ada satu thread pun yang menunggu di titik itu. Pemanggilan join() pada akhirnya setara get() tanpa checked exception, karena exception milik CompletableFuture semuanya tidak diperiksa. Hasil angkanya muncul dua kali karena dua objek CompletableFuture berisi nilai yang sama, satu bertipe Double dan satu lagi bertipe String.

Menggabungkan dua alur dengan thenCombine#

Pekerjaan nyata sering memecah tugas menjadi dua bagian yang berdiri sendiri, lalu menggabungkan hasilnya. Metode thenCombine menerima CompletableFuture lain dan fungsi penggabungnya, dan fungsi itu baru dijalankan ketika kedua alur selesai. Selama menunggu, tidak ada thread yang menganggur khusus untuk keduanya, karena penggabungan hanya terdaftar sebagai kelanjutan (continuation) dari dua hasil tersebut.

import java.util.concurrent.CompletableFuture;

public class DemoThenCombine {

    public static void main(String[] args) {
        String kalimat = "belajar konkurensi di java";

        CompletableFuture<Integer> jumlahKata = CompletableFuture
                .supplyAsync(() -> kalimat.split(" ").length);
        CompletableFuture<Integer> jumlahKarakter = CompletableFuture
                .supplyAsync(() -> kalimat.length());

        int gabungan = jumlahKata
                .thenCombine(jumlahKarakter, (kata, karakter) -> kata + karakter)
                .join();

        System.out.println("Jumlah kata     : " + jumlahKata.join());
        System.out.println("Jumlah karakter : " + jumlahKarakter.join());
        System.out.println("Gabungan        : " + gabungan);
    }
}

Jumlah kata     : 4
Jumlah karakter : 26
Gabungan        : 30

Dua perhitungan sederhana di atas mewakili pola yang sesungguhnya, yaitu dua sumber data yang saling lepas dan digabungkan saat keduanya tiba. Bila salah satu alur selesai lebih dulu, hasilnya menunggu di memori tanpa memegang thread apa pun, dan penggabungnya baru dieksekusi setelah alur kedua tiba. Keluarga metodenya cukup lengkap, thenApply untuk satu alur, thenCombine untuk dua alur, dan allOf untuk sekelompok alur yang ditunggu sekaligus.

// tidak-dijalankan
CompletableFuture<Void> semua = CompletableFuture.allOf(alurSatu, alurDua, alurTiga);
semua.join(); // kembali setelah ketiganya selesai, tanpa memegang hasil

Jalur error dengan exceptionally#

Gagalnya tugas asinkron tidak melempar exception pada tempat kejadiannya, melainkan tersimpan di dalam CompletableFuture dan mengalir ke tahap-tahap setelahnya. Metode exceptionally menyediakan jalur cadangan, yaitu fungsi yang baru dipanggil bila alurnya gagal, sedangkan alur yang berhasil melintasinya tanpa menyentuh fungsi itu. Pada demo berikut, pembagian nol membuat alur gagal, lalu jalur cadangannya menyelamatkan hasil akhirnya.

import java.util.concurrent.CompletableFuture;

public class DemoError {

    public static void main(String[] args) {
        CompletableFuture<String> aman = CompletableFuture
                .supplyAsync(() -> 100 / 0)
                .thenApply(hasil -> "Hasil pembagian: " + hasil)
                .exceptionally(t -> "Gagal dihitung: " + t.getMessage());

        System.out.println(aman.join());
    }
}

Gagal dihitung: java.lang.ArithmeticException: / by zero

Exception aslinya adalah ArithmeticException, tetapi pesan yang tercetak memuat nama kelasnya. Penyebabnya adalah CompletableFuture yang membungkus exception tugas dalam CompletionException, dan getMessage-nya mengembalikan isi pembungkus itu. Tahap thenApply yang berada di antara sumber error dan jalur cadangan dilewati begitu saja, sebab alur yang gagal tidak lagi memproses nilai, melainkan meneruskan kegagalannya sampai ditemukan penangan. Metode handle menawarkan bentuk yang serupa dengan satu fungsi untuk dua keadaan, berhasil maupun gagal, bila jalurnya ingin ditulis dalam satu tempat.

Kesalahan umum#

• Memanggil join() atau get() tepat setelah setiap tahap. Alur yang kembali berurutan itu bukan lagi asinkron, karena main menunggu di setiap titik. Susun seluruh alurnya dahulu, ambil hasilnya di ujung.
• Membiarkan alur gagal tanpa penangan. join() pada alur gagal melempar CompletionException yang menyimpan penyebabnya, dan traceback-nya sulit ditelusuri. Pasang exceptionally atau handle.
• Menjalankan tugas yang memblokir pada common pool. Operasi menunggu jaringan di dalam supplyAsync tanpa executor khusus menyita thread milik JVM yang dipakai bersama. Berikan executor sendiri untuk tugas yang menunggu lama.
• Mengira tahap thenApply selalu berjalan di thread lain. Bila alurnya sudah selesai ketika thenApply dipanggil, fungsinya dieksekusi langsung oleh thread pemanggil. Perilaku itu sah, tetapi perlu diingat saat menulis kode yang menyentuh keadaan bersama.

Ringkasan#

• CompletableFuture menyusun pekerjaan asinkron sebagai rangkaian kelanjutan, sehingga tidak ada thread yang menunggu sambil berdiri.
• supplyAsync memulai alur, thenApply mengubah hasilnya, dan thenCombine menggabungkan dua alur yang saling lepas.
• Exception tugas dibungkus CompletionException dan mengalir sampai ditemukan exceptionally atau handle.
• join() setara get() tanpa checked exception, dan sebaiknya dipanggil sekali di ujung alur.

Referensi#

• Class CompletableFuture — Oracle — dokumentasi resmi keluarga thenApply, thenCombine, allOf, dan exceptionally.
• Interface CompletionStage — Oracle — kontrak kelanjutan yang menjadi dasar CompletableFuture.