Prosesor modern menjalankan banyak thread secara sungguhan sekaligus, dan stream menawarkan pintu masuk termurah ke dunia itu: cukup panggil parallelStream() atau parallel(), maka pipeline yang sudah kita tulis dipecah dan dijalankan serentak di beberapa inti. Tawaran ini perlu disikapi hati-hati — paralelisasi bukan tombol yang selalu mempercepat, dan satu lambda yang tidak disiplin bisa membuat hasil pipeline salah secara senyap. Pada Chapter ini kita pelajari mekanisme pembagian kerjanya, membuktikan hasil yang benar dan terurut lewat collect, lalu menutup dengan daftar kapan paralel layak dipakai.

Cara Kerja Stream Paralel #

Stream paralel dibangun di atas kerangka fork/join. Saat terminal operation dipanggil, Spliterator sumber data memanggil trySplit() berulang kali — setiap pemanggilan membelah data menjadi dua potong yang kira-kira sama besar sampai potongannya terlalu kecil untuk dibelah lagi. Setiap potongan diproses oleh satu worker sebagai pipeline sekuensial yang utuh, lalu hasil antar potongan digabungkan oleh combiner milik terminal operation.

Seluruh worker berenang di ForkJoinPool.commonPool() — kumpulan thread bersama seluruh aplikasi, yang jumlahnya mengikuti jumlah inti prosesor dikurangi satu (thread pemanggil ikut bekerja sehingga tidak ada inti yang menganggur). Karena pool ini dibagikan, pipeline paralel yang berat bisa berebut worker dengan komponen lain di aplikasi yang sama. Cara mengaktifkannya sendiri sederhana: list.parallelStream() sejak awal, atau stream.parallel() di tengah pipeline — flag paralel bisa juga ditarik kembali dengan sequential().

Pipeline Paralel dengan Hasil Terurut #

Kekhawatiran pertama pemula biasanya urutan: kalau data diolah serentak, apakah hasil collect ikut teracak? Tidak — collect pada stream paralel ORDERED menggabungkan potongan-potongan hasil sesuai urutan sumber, seperti sudah kita bahas pada Chapter Stream Ordering. Program berikut menyaring 10.000 produk secara paralel dan memeriksa tepi hasilnya:

java ParalelProduk.java 
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ParalelProduk {

    record Produk(String nama, String kategori, int harga) {}

    public static void main(String[] args) {
        // 10.000 produk dibangkitkan secara deterministik
        List<Produk> etalase = IntStream.rangeClosed(1, 10_000)
            .mapToObj(n -> new Produk("Produk-" + n,
                n % 2 == 0 ? "elektronik" : "atk", n * 100))
            .collect(Collectors.toList());

        // Pipeline paralel: hasil tetap terurut berkat collect
        List<String> elektronikMurah = etalase.parallelStream()
            .filter(p -> p.kategori().equals("elektronik"))
            .filter(p -> p.harga() <= 250_000)
            .map(p -> p.nama() + " (Rp" + p.harga() + ")")
            .collect(Collectors.toList());

        System.out.println("jumlah hasil   = " + elektronikMurah.size());
        System.out.println("hasil pertama  = " + elektronikMurah.get(0));
        System.out.println("hasil terakhir = " + elektronikMurah.get(elektronikMurah.size() - 1));
    }
}

Output yang Dihasilkan:

text 
jumlah hasil   = 1250
hasil pertama  = Produk-2 (Rp200)
hasil terakhir = Produk-2500 (Rp250000)

Mari kita telaah baris-baris kode di atas:

  1. Baris 11-14 (IntStream.rangeClosed(1, 10_000).mapToObj(...)): sumber data dibangkitkan deterministik — produk ke-n punya kategori bergantian dan harga n * 100 — sehingga jawaban benar bisa dihitung di luar kepala: kategori elektronik jatuh pada n genap, dan harga di bawah atau sama dengan Rp250.000 berarti n sampai 2500.
  2. Baris 17 (etalase.parallelStream()): satu kata ini mengubah seluruh pipeline menjadi paralel. filter, map, dan collect di bawahnya ditulis persis seperti versi sekuensial; tidak ada kode konkurensi yang perlu kita tulis manual.
  3. Baris 18-19 (dua filter berantai): tiap potongan data memfilter sendiri produk elektronik dengan harga di bawah batas. Kedua filter bebas efek samping sehingga aman dijalankan serentak oleh banyak thread.
  4. Baris 21 (collect(Collectors.toList())): tiap worker mengumpulkan potongan daftarnya, lalu combiner menyusun potongan-potongan itu menurut urutan sumber. Ukuran 1250 (bilangan genap 2 sampai 2500), hasil pertama Produk-2 (Rp200), dan hasil terakhir Produk-2500 (Rp250000) — semuanya terurut dan bisa diverifikasi tanpa melihat urutan eksekusi thread.

Jawaban yang Identik antara Sekuensial dan Paralel #

Kebenaran hasil paralel bersandar pada dua sifat operasi: asosiatif (pengelompokan perhitungan tidak mengubah jawaban, seperti penjumlahan) dan bebas efek samping (lambda tidak menyentuh keadaan bersama). Program berikut membandingkan jawaban sekuensial dan paralel untuk sum dan reduce:

java PerbandinganHasil.java 
import java.util.stream.IntStream;

public class PerbandinganHasil {

    public static void main(String[] args) {
        int batas = 1_000_000;

        // sum: operasi asosiatif, paralel tetap menjawab sama.
        // mapToLong menghindari overflow karena total melewati rentang int.
        long jumlahSekuensial = IntStream.rangeClosed(1, batas)
            .mapToLong(n -> (long) n * 3).sum();
        long jumlahParalel = IntStream.rangeClosed(1, batas).parallel()
            .mapToLong(n -> (long) n * 3).sum();
        System.out.println("sekuensial = " + jumlahSekuensial);
        System.out.println("paralel    = " + jumlahParalel);
        System.out.println("identik    = " + (jumlahSekuensial == jumlahParalel));

        // reduce: identitas nol + Integer::sum yang asosiatif
        int reduceSekuensial = IntStream.rangeClosed(1, 1000)
            .reduce(0, Integer::sum);
        int reduceParalel = IntStream.rangeClosed(1, 1000).parallel()
            .reduce(0, Integer::sum);
        System.out.println("reduce sekuensial = " + reduceSekuensial);
        System.out.println("reduce paralel    = " + reduceParalel);
        System.out.println("reduce identik    = " + (reduceSekuensial == reduceParalel));
    }
}

Output yang Dihasilkan:

text 
sekuensial = 1500001500000
paralel    = 1500001500000
identik    = true
reduce sekuensial = 500500
reduce paralel    = 500500
reduce identik    = true

Mari kita telaah baris-baris kode di atas:

  1. Baris 10-13 (mapToLong(n -> (long) n * 3).sum()): total 3 × (1 + 2 + … + 1.000.000) mencapai 1.500.001.500.000 — melewati rentang int, sehingga mapToLong dipakai agar tidak terjadi integer overflow. Karena penjumlahan asosiatif, sekuensial dan paralel menghasilkan angka yang sama persis meskipun urutan penjumlahannya berbeda.
  2. Baris 12 dan 21 (.parallel() di tengah pipeline): flag paralel bisa dinyalakan kapan saja sebelum terminal operation; posisi setelah rangeClosed atau sebelum reduce tidak mengubah artinya.
  3. Baris 19-22 (reduce(0, Integer::sum)): reduce paralel butuh identitas yang benar (0 untuk penjumlahan) dan combiner asosiatif (Integer::sum). Kedua syarat terpenuhi, jawaban keduanya 500500 — jumlah 1 sampai 1000.

Bahaya Lambda Stateful #

Pola yang paling sering menjatuhkan pipeline paralel adalah menulis ke koleksi luar dari dalam lambda. Kode berikut tampak wajar pada stream sekuensial dan gagal diam-diam pada stream paralel:

java Contoh lambda stateful yang salah 
List<Integer> hasil = new ArrayList<>();          // tidak aman untuk thread

List<Integer> data = List.of(1, 2, 3, 4, 5, 6, 7, 8);
data.parallelStream()
    .filter(n -> n % 2 == 0)
    .forEach(n -> hasil.add(n));                  // dua thread menulis bersamaan

ArrayList tidak dirancang untuk dimodifikasi banyak thread sekaligus. Dua worker bisa menulis pada posisi yang sama sehingga satu nilai tertimpa, atau ukuran internalnya tidak sinkron sehingga ArrayIndexOutOfBoundsException muncul sewaktu-waktu — dan pada data kecil program kadang tampak benar, membuat kesalahan ini sulit ditemukan. Versi yang benar tidak menulis apa pun ke luar: cukup .collect(Collectors.toList()), dan mesin stream yang mengurus penggabungan potongan dengan aman.

Kapan Paralel Membantu dan Kapan Tidak #

Paralelisasi menang besar hanya bila keempat syarat berikut terpenuhi bersamaan:

  • Pekerjaan berat secara CPU: lambda melakukan komputasi, bukan menunggu I/O. Stream yang membaca berkas atau memanggil layanan jaringan justru memblokir worker di common pool dan mengganggu seluruh aplikasi.
  • Ukuran data besar: pembelahan data, pengiriman tugas, dan penggabungan hasil punya harga tetap. Pada ratusan elemen, biaya koordinasi biasa melampaui penghematan waktu komputasi.
  • Sumber mudah dibagi: ArrayList, array, dan IntStream.range membelah rapi; LinkedList atau iterator dari iterate sulit dipecah sehingga paralelisasinya timpang.
  • Operasi asosiatif dan bebas efek samping: sum, count, joining, dan collect ke toList aman; mutasi koleksi luar dan forEach yang mencetak sambil mengandalkan urutan tidak aman.

Untuk kasus di luar daftar itu — terutama data kecil, I/O, atau kebutuhan urutan ketat pada forEach — pipeline sekuensial lebih cepat, lebih sederhana, dan lebih mudah diperiksa. Keputusan memakai paralel sebaiknya diambil dari hasil pengukuran, bukan asumsi; cara mengukur yang benar tanpa terjebak angka waktu yang fluktuatif menjadi bahasan Chapter Performa Stream berikutnya.

Referensi #

Berkas Sumber

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

ParalelProduk.java

Unduh berkas
java 
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ParalelProduk {

    record Produk(String nama, String kategori, int harga) {}

    public static void main(String[] args) {
        // 10.000 produk dibangkitkan secara deterministik
        List<Produk> etalase = IntStream.rangeClosed(1, 10_000)
            .mapToObj(n -> new Produk("Produk-" + n,
                n % 2 == 0 ? "elektronik" : "atk", n * 100))
            .collect(Collectors.toList());

        // Pipeline paralel: hasil tetap terurut berkat collect
        List<String> elektronikMurah = etalase.parallelStream()
            .filter(p -> p.kategori().equals("elektronik"))
            .filter(p -> p.harga() <= 250_000)
            .map(p -> p.nama() + " (Rp" + p.harga() + ")")
            .collect(Collectors.toList());

        System.out.println("jumlah hasil   = " + elektronikMurah.size());
        System.out.println("hasil pertama  = " + elektronikMurah.get(0));
        System.out.println("hasil terakhir = " + elektronikMurah.get(elektronikMurah.size() - 1));
    }
}

PerbandinganHasil.java

Unduh berkas
java 
import java.util.stream.IntStream;

public class PerbandinganHasil {

    public static void main(String[] args) {
        int batas = 1_000_000;

        // sum: operasi asosiatif, paralel tetap menjawab sama.
        // mapToLong menghindari overflow karena total melewati rentang int.
        long jumlahSekuensial = IntStream.rangeClosed(1, batas)
            .mapToLong(n -> (long) n * 3).sum();
        long jumlahParalel = IntStream.rangeClosed(1, batas).parallel()
            .mapToLong(n -> (long) n * 3).sum();
        System.out.println("sekuensial = " + jumlahSekuensial);
        System.out.println("paralel    = " + jumlahParalel);
        System.out.println("identik    = " + (jumlahSekuensial == jumlahParalel));

        // reduce: identitas nol + Integer::sum yang asosiatif
        int reduceSekuensial = IntStream.rangeClosed(1, 1000)
            .reduce(0, Integer::sum);
        int reduceParalel = IntStream.rangeClosed(1, 1000).parallel()
            .reduce(0, Integer::sum);
        System.out.println("reduce sekuensial = " + reduceSekuensial);
        System.out.println("reduce paralel    = " + reduceParalel);
        System.out.println("reduce identik    = " + (reduceSekuensial == reduceParalel));
    }
}