Parallel Streams dan Concurrency

Section Stream API sudah membahas parallel stream dari sisi pipeline: Spliterator, syarat asosiatif, dan jebakan lambda stateful. Bab ini mengambil sudut yang berbeda — sisi mesin konkurensi yang berada di bawahnya. Satu baris .parallel() memindahkan seluruh beban ke ForkJoinPool.commonPool(), sebuah pool bersama seluruh JVM yang diam-diam juga dipakai komponen lain. Konsekuensinya besar: stream paralel yang salah urusan tidak merusak hasil komputasinya sendiri, tetapi bisa melumpuhkan bagian aplikasi lain yang kebetulan menumpang di pool yang sama. Kita akan membuktikan pembagian pool tersebut, membedah skenario bahaya pemblokiran worker, lalu mempelajari dua cara mengendalikan ukurannya.

Satu Pool untuk Seluruh Aplikasi #

ForkJoinPool.commonPool() diciptakan malas saat pertama kali dibutuhkan, dengan jumlah pekerja tetap: banyaknya inti CPU dikurangi satu (minimal satu). Thread yang memanggil terminal operation ikut bekerja sehingga seluruh inti termanfaatkan. Dua penyewa utama pool ini adalah parallel stream dan CompletableFuture.supplyAsync() tanpa argumen executor — dua subsistem yang sama sekali tidak saling kenal tetapi berbagi tempat kerja:

import java.util.concurrent.CompletableFuture;
import java.util.stream.LongStream;

public class KolamBersama {
    public static void main(String[] args) {
        // CompletableFuture tanpa executor kustom berjalan di
        // ForkJoinPool.commonPool() -- pool bersama seluruh JVM.
        CompletableFuture<String> sapaan = CompletableFuture.supplyAsync(
                () -> "halo dari completablefuture");

        // Parallel stream memakai pool yang sama persis.
        long jumlah = LongStream.rangeClosed(1, 1_000_000)
                .parallel()
                .map(n -> n * n)
                .sum();

        String hasilSapaan = sapaan.join();
        System.out.println("hasil stream paralel     : " + jumlah);
        System.out.println("hasil completablefuture  : " + hasilSapaan);
        System.out.println("kedua jalur tuntas       : " + sapaan.isDone());
    }
}

Output yang Dihasilkan:

hasil stream paralel     : 333333833333500000
hasil completablefuture  : halo dari completablefuture
kedua jalur tuntas       : true

Mari kita bedah apa yang sebenarnya terjadi:

1. supplyAsync tanpa executor: tahap pertama CompletableFuture ini dijadwalkan ke commonPool. Sejak Java 9, commonPool juga memakai virtual thread event handler untuk keperluan internal, tetapi pekerja utamanya tetap thread platform berjumlah tetap.
2. LongStream...parallel().map(...).sum(): terminal operation sum() membelah rentang lewat trySplit() lalu menyerahkan potongan-potongan itu ke pool yang sama. Kedua pekerjaan — aliran dan future — kini berebut pekerja yang sama.
3. Hasil kedua jalur tetap benar: 333333833333500000 (rumus jumlah kuadrat) dan teks sapaan utuh. Interferensi di commonPool berupa perlambatan, bukan kerusakan hasil — thread pemanggil ikut membantu sehingga keduanya pasti tuntas. Baris ketiga (isDone() setelah join()) membuktikan future selesai sebelum pencetakan.

Pembagian satu pool adalah keputusan desain yang sadar: pool kecil hemat memori dan cukup untuk komputasi murni. Masalahnya muncul ketika penghuni pool tidak lagi berperilaku seperti yang diasumsikan desainnya — komputasi singkat dan tanpa penungguan.

Saat Parallel Stream Menjadi Beban #

CommonPool didesain dengan asumsi pekerjanya mengerjakan komputasi CPU-bound yang tidak pernah menunggu. Asumsi itu runtuh begitu lambda di dalam stream memanggil operasi pemblokir:

{title="Pola yang menunggu di dalam stream paralel (jangan ditiru)"}
List<Pesanan> semua = muatSemuaPesanan();

// Setiap elemen "memproses" dengan memanggil layanan luar yang lambat.
List<String> status = semua.parallelStream()
        .map(p -> panggilLayananEksternal(p))  // menunggu jaringan 200-500 ms
        .collect(Collectors.toList());

// Di bagian lain aplikasi, pada saat yang sama:
CompletableFuture<String> laporan =
        CompletableFuture.supplyAsync(() -> susunLaporan()); // menunggu di antrean

Mekanisme melumpuhkannya bertingkat dan sunyi. Pada mesin berinti N, commonPool hanya punya N−1 pekerja; pipeline di atas menahan setiap pekerja berpuluh-puluh milidetik dalam keadaan menganggur menunggu jaringan. Bila jumlah elemen melebihi jumlah pekerja, seluruh pekerja tersedot ke tugas menunggu — dan setiap CompletableFuture.supplyAsync yang datang kemudian menginap di antrean pool tanpa thread yang menjalankannya. Tidak ada exception, tidak ada log galat: seluruh subsistem asinkron aplikasi berhenti bereaksi sampai panggilan jaringan selesai satu per satu.

Aturan praktisnya satu kalimat: parallel stream untuk komputasi, bukan untuk penungguan. Untuk beban I/O, alat yang tepat bukan commonPool melainkan executor dengan virtual threads — topik dua chapter setelah ini. Lambda stateful dan efek samping sudah dibahas pada Section Stream API, dan keduanya memperparah situasi yang sama: hasil tak stabil dan worker terpakai untuk pekerjaan serial yang tidak terdistribusi.

Mengendalikan Ukuran Pool #

Dua kendali tersedia, masing-masing dengan cakupan berbeda. Yang pertama adalah properti sistem java.util.concurrent.ForkJoinPool.common.parallelism — di-set saat JVM start dan berlaku untuk seluruh commonPool:

java -Djava.util.concurrent.ForkJoinPool.common.parallelism=4 AplikasiPembayaran

Properti ini mengubah ukuran pool bersama, bukan pool khusus satu pipeline; menggesernya memengaruhi semua penghuni sekaligus, sehingga jarang menjadi jawaban yang tepat untuk masalah satu stream. Kendali kedua jauh lebih tertarget: jalankan pipeline di dalam tugas milik ForkJoinPool buatan sendiri:

import java.util.concurrent.ForkJoinPool;
import java.util.stream.LongStream;

public class KolamKhusus {
    public static void main(String[] args) throws Exception {
        // Teknik kendali ukuran pool: jalankan pipeline paralel DI DALAM
        // tugas milik ForkJoinPool khusus. Terminal operation stream akan
        // memakai pekerja pool ini, bukan commonPool bersama.
        ForkJoinPool pool = new ForkJoinPool(2);
        try {
            long jumlah = pool.submit(() ->
                    LongStream.rangeClosed(1, 1_000_000)
                            .parallel()
                            .map(n -> n * n)
                            .sum())
                    .get();
            System.out.println("hasil di pool khusus (2 pekerja): " + jumlah);
        } finally {
            pool.shutdown();
        }
    }
}

Output yang Dihasilkan:

hasil di pool khusus (2 pekerja): 333333833333500000

Tiga hal yang perlu dicermati dari teknik ini:

1. pool.submit(...) membungkus terminal operation: sum() dijalankan di dalam ForkJoinTask milik pool khusus, dan pembelahan-pembelahan stream di bawahnya menumpang pool yang sama. Hasil 333333833333500000 identik dengan versi commonPool — berganti pool tidak mengubah matematika, hanya tempat kerjanya.
2. get() pada hasil submit: pemanggil menunggu di luar pool, sehingga dua pekerja pool bebas mengerjakan potongan stream tanpa thread pemanggil yang ikut membantu.
3. Status teknik ini: idiom submit-ke-pool-khusus dikenal luas dan bekerja pada seluruh JDK modern, tetapi bergantung pada detail implementasi, bukan jaminan spesifikasi — perlakukan sebagai solusi praktis yang perlu diuji ulang saat upgrade JDK besar.

Panduan Memilih Alat #

Pola yang berulang pada tabel di atas: beban penungguan menuntut thread yang murah. Java 21 menjawab kebutuhan itu dengan virtual threads — thread yang dikelola JVM, bukan sistem operasi, sehingga jutaan penunggu dapat hidup bersamaan. Chapter berikutnya membedahnya sampai ke carrier thread-nya.

Referensi #

• Oracle: Class ForkJoinPool (Java API Specification) — commonPool
• Oracle: Class CompletableFuture (Java API Specification)
• dev.java — Parallel streams considerations

import java.util.concurrent.CompletableFuture;
import java.util.stream.LongStream;

public class KolamBersama {
    public static void main(String[] args) {
        // CompletableFuture tanpa executor kustom berjalan di
        // ForkJoinPool.commonPool() -- pool bersama seluruh JVM.
        CompletableFuture<String> sapaan = CompletableFuture.supplyAsync(
                () -> "halo dari completablefuture");

        // Parallel stream memakai pool yang sama persis.
        long jumlah = LongStream.rangeClosed(1, 1_000_000)
                .parallel()
                .map(n -> n * n)
                .sum();

        String hasilSapaan = sapaan.join();
        System.out.println("hasil stream paralel     : " + jumlah);
        System.out.println("hasil completablefuture  : " + hasilSapaan);
        System.out.println("kedua jalur tuntas       : " + sapaan.isDone());
    }
}

import java.util.concurrent.ForkJoinPool;
import java.util.stream.LongStream;

public class KolamKhusus {
    public static void main(String[] args) throws Exception {
        // Teknik kendali ukuran pool: jalankan pipeline paralel DI DALAM
        // tugas milik ForkJoinPool khusus. Terminal operation stream akan
        // memakai pekerja pool ini, bukan commonPool bersama.
        ForkJoinPool pool = new ForkJoinPool(2);
        try {
            long jumlah = pool.submit(() ->
                    LongStream.rangeClosed(1, 1_000_000)
                            .parallel()
                            .map(n -> n * n)
                            .sum())
                    .get();
            System.out.println("hasil di pool khusus (2 pekerja): " + jumlah);
        } finally {
            pool.shutdown();
        }
    }
}