Fork/Join Framework

Executor yang dibahas bab-bab lalu unggul untuk task yang datang dari luar: permintaan web, pekerjaan antrean. Ada satu bentuk beban lain yang justru lahir dari dalam: komputasi rekursif — menghitung seluruh array, memproses pohon keputusan, menganalisis rentang data — yang pekerjaannya bisa dibelah berulang kali hingga cukup kecil. Membelah pekerjaan seperti itu ke ExecutorService biasa berisiko dua masalah: puluhan ribu tugas kecil membanjiri antrean, dan thread yang menunggu hasil tugas turunan (lewat get()) duduk menganggur sambil memegang thread mahal. Framework Fork/Join didesain khusus untuk bentuk beban ini: setiap pekerja membelah tugasnya sendiri, mengantrekan belahan, dan — kunci keunggulannya — mencuri pekerjaan dari antrean pekerja lain yang masih sibuk.

RecursiveTask: Membelah dan Menggabungkan #

Dua kelas dasar framework ini adalah RecursiveTask<V> (menghasilkan nilai) dan RecursiveAction (hanya efek samping). Keduanya menuntut method compute() yang memuat keputusan yang sama: bila pekerjaan sudah di bawah ambang, kerjakan langsung; bila masih besar, belah dua. Program berikut menjumlahkan bilangan 1 sampai 10.000 lewat pembelahan bertingkat:

java
PenjumlahanForkJoin.java
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class PenjumlahanForkJoin {
    private static final int AMBANG = 1_000; // ukuran maksimum dihitung langsung

    // RecursiveTask<Long>: tugas rekursif yang mengembalikan nilai.
    private static class TugasJumlah extends RecursiveTask<Long> {
        private final int dari;
        private final int sampai;

        TugasJumlah(int dari, int sampai) {
            this.dari = dari;
            this.sampai = sampai;
        }

        @Override
        protected Long compute() {
            if (sampai - dari + 1 <= AMBANG) {
                long total = 0; // cukup kecil: jumlahkan berurutan
                for (int i = dari; i <= sampai; i++) {
                    total += i;
                }
                return total;
            }
            int tengah = (dari + sampai) / 2;
            TugasJumlah kiri = new TugasJumlah(dari, tengah);
            TugasJumlah kanan = new TugasJumlah(tengah + 1, sampai);
            kiri.fork();                     // antrekan paruh kiri ke pool
            long hasilKanan = kanan.compute(); // kerjakan paruh kanan sendiri
            long hasilKiri = kiri.join();      // tunggu dan ambil hasil kiri
            return hasilKiri + hasilKanan;
        }
    }

    public static void main(String[] args) throws Exception {
        long berurutan = 0; // patokan: penjumlahan satu thread
        for (int i = 1; i <= 10_000; i++) {
            berurutan += i;
        }
        System.out.println("penjumlahan berurutan : " + berurutan);

        ForkJoinPool pool = new ForkJoinPool();
        try {
            // invoke: kirim tugas teratas dan tunggu sampai selesai.
            long hasilInvoke = pool.invoke(new TugasJumlah(1, 10_000));
            System.out.println("fork/join via invoke  : " + hasilInvoke);

            // submit: kirim dulu, ambil hasilnya nanti lewat get().
            ForkJoinTask<Long> tugas = pool.submit(new TugasJumlah(1, 10_000));
            System.out.println("fork/join via submit  : " + tugas.get());
        } finally {
            pool.shutdown();
        }
    }
}

Output yang Dihasilkan:

text
penjumlahan berurutan : 50005000
fork/join via invoke  : 50005000
fork/join via submit  : 50005000

Mari kita bedah inti polanya:

sampai - dari + 1 <= AMBANG: keputusan berhenti membelah. Rentang 10.000 dibelah bertingkat menjadi 10 rentang @1.000; tiap rentang dijumlahkan berurutan oleh satu thread. Ambang yang tepat menyeimbangkan biaya tugas versus biaya pengelolaan tugas — terlalu kecil berarti ribuan tugas mikro yang lebih banyak waktu habis untuk administrasi daripada komputasi.
kiri.fork() lalu kanan.compute(): urutan ini bukan kebetulan. Paruh kiri diantrekan dulu, lalu thread yang sama langsung mengerjakan paruh kanan — tidak menunggu, tidak memblokir. Thread hanya berhenti sesaat di kiri.join(), dan pada titik itu pekerjaan tersisa sudah seminimal mungkin.
invoke versus submit: invoke(task) mengirim tugas teratas dan menunggu selesai dalam satu panggilan; submit(task) mengembalikan ForkJoinTask (turunan Future) sehingga pemanggil bisa mengerjakan hal lain sebelum get(). Keduanya menghasilkan nilai yang sama persis.
Hasil 50005000 di ketiga baris: rumus 10.000 × 10.001 / 2 — patokan berurutan dan dua jalur fork/join sepakat; pembelahan tidak mengubah hasil penjumlahan.

Work Stealing: Mengapa Antreannya Berbeda #

Tiap pekerja ForkJoinPool memiliki antrean ganda ujung (deque) sendiri. Pemilik antrean mengambil tugas dari satu ujung, sementara pekerja lain yang kehabisan tugas mengambil dari ujung yang berlawanan. Dua arah itu meminimalkan bentrokan: pemilik mengerjakan tugas termuda (hasil belahan terakhir, paling besar kemungkinan masih bisa dibelah lagi), pencuri membawa tugas tertua (paling besar, paling bernilai dicuri).

flowchart TD A[Tugas 1..10000 masuk ke pekerja-1] --> B[pekerja-1 membelah: kiri di-fork, kanan dihitung] B --> C[antrean pekerja-1 menumpuk belahan] C --> D{pekerja-2 kehabisan tugas?} D -- ya --> E[pekerja-2 mencuri dari ujung berlawanan] E --> F[pekerja-2 membelah hasil curian sendiri] D -- tidak --> G[pekerja-2 melanjutkan antreannya sendiri]

Efek bersihnya: keseimbangan beban terjadi otomatis tanpa penjadwal pusat. Pekerja yang kebetulan menerima tugas mudah selesai lebih cepat lalu langsung membantu pekerja yang tersangkut di cabang berat. ForkJoinPool bawaan memakai jumlah pekerja sebanyak inti CPU; berbeda dari newFixedThreadPool, penambahan pekerja di atas jumlah inti tidak membantu komputasi murni seperti ini.

RecursiveAction dan invokeAll #

Untuk pekerjaan tanpa nilai balik, RecursiveAction menjadi induknya; hasil ditanam pada efek sampingnya. Pola kedua yang layak dikenal: invokeAll(kiri, kanan) — mem-fork kedua belahan lalu menunggu keduanya, lebih ringkas daripada pasangan fork/compute/join:

java
AksiKaliDua.java
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class AksiKaliDua {
    private static final int AMBANG = 1_000;

    // RecursiveAction: tugas rekursif TANPA nilai balik -- hasil
    // tersimpan pada efek sampingnya (array yang digandakan).
    private static class AksiGandakan extends RecursiveAction {
        private final int[] data;
        private final int dari;
        private final int sampai; // batas atas eksklusif

        AksiGandakan(int[] data, int dari, int sampai) {
            this.data = data;
            this.dari = dari;
            this.sampai = sampai;
        }

        @Override
        protected void compute() {
            if (sampai - dari <= AMBANG) {
                for (int i = dari; i < sampai; i++) {
                    data[i] *= 2;
                }
                return;
            }
            int tengah = (dari + sampai) / 2;
            AksiGandakan kiri = new AksiGandakan(data, dari, tengah);
            AksiGandakan kanan = new AksiGandakan(data, tengah, sampai);
            invokeAll(kiri, kanan); // fork keduanya, tunggu keduanya selesai
        }
    }

    public static void main(String[] args) {
        int[] data = new int[8_000];
        for (int i = 0; i < data.length; i++) {
            data[i] = i + 1;
        }
        long sebelum = 0;
        for (int nilai : data) {
            sebelum += nilai;
        }

        ForkJoinPool pool = new ForkJoinPool();
        try {
            pool.invoke(new AksiGandakan(data, 0, data.length));
        } finally {
            pool.shutdown();
        }

        long sesudah = 0;
        for (int nilai : data) {
            sesudah += nilai;
        }
        System.out.println("jumlah sebelum         : " + sebelum);
        System.out.println("jumlah sesudah         : " + sesudah);
        System.out.println("semua elemen digandakan: " + (sesudah == sebelum * 2));
    }
}

Output yang Dihasilkan:

text
jumlah sebelum         : 32004000
jumlah sesudah         : 64008000
semua elemen digandakan: true

Dua catatan untuk aksi tersebut:

invokeAll(kiri, kanan): kedua belahan dikirim lalu ditunggu bersamaan — thread pemanggil ikut membantu mengerjakan salah satunya, jadi tidak ada waktu tunggu menganggur. Pilihan gaya antara fork/compute/join dan invokeAll adalah selera keterbacaan; keduanya sama-sama efisien.
Pembuktian lewat agregat: elemen diperiksa satu per satu tidak akan stabil dicetak (urutan thread bebas), sehingga buktinya adalah jumlah seluruh array: 32.004.000 menjadi 64.008.000 — tepat dua kali lipat.

ForkJoinPool.commonPool #

Membuat pool sendiri tidak selalu perlu. Satu instance bersama — ForkJoinPool.commonPool() — tersedia untuk seluruh JVM dengan jumlah pekerja jumlah inti - 1 (minimal satu). Pool inilah yang diam-diam dipakai CompletableFuture.supplyAsync() tanpa argumen executor dan, seperti dibahas pada Chapter berikutnya, seluruh parallel stream. Mengetahui keberadaannya penting: kode yang menghabiskan common pool dengan tugas lambat ikut melumpuhkan parallel stream di bagian aplikasi lain.

Fork/join memberi kendali penuh atas pembelahan; namun untuk komputasi atas koleksi, Java menyediakan sintaks yang jauh lebih ringkas yang berjalan di atas mesin yang sama — parallel stream — beserta jebakan-jebakan konkurensinya.

Referensi #

Berkas Sumber

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

java
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class AksiKaliDua {
    private static final int AMBANG = 1_000;

    // RecursiveAction: tugas rekursif TANPA nilai balik -- hasil
    // tersimpan pada efek sampingnya (array yang digandakan).
    private static class AksiGandakan extends RecursiveAction {
        private final int[] data;
        private final int dari;
        private final int sampai; // batas atas eksklusif

        AksiGandakan(int[] data, int dari, int sampai) {
            this.data = data;
            this.dari = dari;
            this.sampai = sampai;
        }

        @Override
        protected void compute() {
            if (sampai - dari <= AMBANG) {
                for (int i = dari; i < sampai; i++) {
                    data[i] *= 2;
                }
                return;
            }
            int tengah = (dari + sampai) / 2;
            AksiGandakan kiri = new AksiGandakan(data, dari, tengah);
            AksiGandakan kanan = new AksiGandakan(data, tengah, sampai);
            invokeAll(kiri, kanan); // fork keduanya, tunggu keduanya selesai
        }
    }

    public static void main(String[] args) {
        int[] data = new int[8_000];
        for (int i = 0; i < data.length; i++) {
            data[i] = i + 1;
        }
        long sebelum = 0;
        for (int nilai : data) {
            sebelum += nilai;
        }

        ForkJoinPool pool = new ForkJoinPool();
        try {
            pool.invoke(new AksiGandakan(data, 0, data.length));
        } finally {
            pool.shutdown();
        }

        long sesudah = 0;
        for (int nilai : data) {
            sesudah += nilai;
        }
        System.out.println("jumlah sebelum         : " + sebelum);
        System.out.println("jumlah sesudah         : " + sesudah);
        System.out.println("semua elemen digandakan: " + (sesudah == sebelum * 2));
    }
}

java
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class PenjumlahanForkJoin {
    private static final int AMBANG = 1_000; // ukuran maksimum dihitung langsung

    // RecursiveTask<Long>: tugas rekursif yang mengembalikan nilai.
    private static class TugasJumlah extends RecursiveTask<Long> {
        private final int dari;
        private final int sampai;

        TugasJumlah(int dari, int sampai) {
            this.dari = dari;
            this.sampai = sampai;
        }

        @Override
        protected Long compute() {
            if (sampai - dari + 1 <= AMBANG) {
                long total = 0; // cukup kecil: jumlahkan berurutan
                for (int i = dari; i <= sampai; i++) {
                    total += i;
                }
                return total;
            }
            int tengah = (dari + sampai) / 2;
            TugasJumlah kiri = new TugasJumlah(dari, tengah);
            TugasJumlah kanan = new TugasJumlah(tengah + 1, sampai);
            kiri.fork();                     // antrekan paruh kiri ke pool
            long hasilKanan = kanan.compute(); // kerjakan paruh kanan sendiri
            long hasilKiri = kiri.join();      // tunggu dan ambil hasil kiri
            return hasilKiri + hasilKanan;
        }
    }

    public static void main(String[] args) throws Exception {
        long berurutan = 0; // patokan: penjumlahan satu thread
        for (int i = 1; i <= 10_000; i++) {
            berurutan += i;
        }
        System.out.println("penjumlahan berurutan : " + berurutan);

        ForkJoinPool pool = new ForkJoinPool();
        try {
            // invoke: kirim tugas teratas dan tunggu sampai selesai.
            long hasilInvoke = pool.invoke(new TugasJumlah(1, 10_000));
            System.out.println("fork/join via invoke  : " + hasilInvoke);

            // submit: kirim dulu, ambil hasilnya nanti lewat get().
            ForkJoinTask<Long> tugas = pool.submit(new TugasJumlah(1, 10_000));
            System.out.println("fork/join via submit  : " + tugas.get());
        } finally {
            pool.shutdown();
        }
    }
}