Bagian 1 meninggalkan satu pertanyaan: tahap validasi dan transformasi memproses 30 baris satu per satu, padahal tiap baris tidak saling bergantung. Bagian 2 memparalelkan tahap itu dengan tiga pilar java.util.concurrent: ExecutorService sebagai pemasok thread, CompletableFuture sebagai pengirim task, dan collection konkuren sebagai penampungan hasil. Satu aturan dijaga: pekerja tidak mencetak sendiri hasilnya — hasil dikumpulkan, semua task ditunggu, urutan dipulihkan, baru laporan dicetak oleh main.

Satu Aturan untuk Semua Eksekusi Konkuren #

Galat paling umum pada pipeline paralel bukan deadlock, melainkan keluaran yang berubah antar eksekusi: thread mana selesai lebih dulu, itulah yang cetak lebih dulu. Aturan kita memutus masalah di akarnya: task hanya mengembalikan nilai atau menulis ke collection konkuren; main memanggil gerbang tunggu CompletableFuture.allOf(...).join(); setelah gerbang terbuka, main mencetak hasil dalam urutan yang kita tentukan sendiri — nomor baris. Nama thread, waktu selesai, dan identitas hash tidak pernah masuk keluaran.

Eksekusi Paralel per Baris #

Pipeline kini memiliki jalankanParalel — versi baru dari orkestrasi Bagian 1 yang mengirim satu task per baris ke executor mana pun:

java Pipeline.java 
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/** Pipeline penjualan: baca -> validasi -> transformasi -> agregasi -> tulis. */
public final class Pipeline {

    /** Kategori yang dikenal validator; di luar daftar ini baris ditolak. */
    private static final Set<String> KATEGORI_DIKENAL =
            Set.of("Elektronik", "Pakaian", "Buku", "Peralatan");

    /** Faktor biaya layanan 2% yang dihitung tahap transformasi. */
    private static final BigDecimal BIAYA_LAYANAN = new BigDecimal("1.02");

    private Pipeline() {
        // Kelas utilitas: seluruh anggota static.
    }

    /** Tahap baca: memberi nomor urut pada tiap baris berkas. */
    public static List<BarisData> baca(List<String> isiBerkas) {
        List<BarisData> baris = new ArrayList<>(isiBerkas.size());
        for (int i = 0; i < isiBerkas.size(); i++) {
            baris.add(new BarisData(i + 1, isiBerkas.get(i)));
        }
        return baris;
    }

    /** Tahap validasi lalu transformasi untuk satu baris. */
    public static HasilBaris prosesSatuBaris(BarisData baris) {
        String[] segmen = baris.teks().split(";", -1);
        if (segmen.length != 5) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "jumlah segmen " + segmen.length + ", harus 5");
        }
        LocalDate tanggal;
        try {
            tanggal = LocalDate.parse(segmen[0]);
        } catch (DateTimeParseException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "tanggal tidak terurai: '" + segmen[0] + "'");
        }
        String kota = segmen[1].trim();
        String pelanggan = segmen[2].trim();
        String kategori = segmen[3].trim();
        if (kota.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "kota kosong");
        }
        if (pelanggan.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "nama pelanggan kosong");
        }
        if (!KATEGORI_DIKENAL.contains(kategori)) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "kategori tidak dikenal: '" + kategori + "'");
        }
        BigDecimal nilai;
        try {
            nilai = new BigDecimal(segmen[4].trim());
        } catch (NumberFormatException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai tidak terurai: '" + segmen[4].trim() + "'");
        }
        if (nilai.signum() <= 0) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai harus positif: " + nilai.toPlainString());
        }
        EventPenjualan event = new EventPenjualan(baris.nomor(), tanggal, kota,
                pelanggan, kategori, nilai);
        return new HasilBaris.Sah(ubah(event));
    }

    /** Tahap transformasi: bakukan kota dan hitung nilai berjasa 2%. */
    public static KejadianTerproses ubah(EventPenjualan event) {
        BigDecimal berjasa = event.nilai().multiply(BIAYA_LAYANAN)
                .setScale(2, RoundingMode.HALF_EVEN);
        return new KejadianTerproses(event.nomorBaris(), event.tanggal(),
                event.kota().toUpperCase(Locale.ROOT), event.pelanggan(),
                event.kategori(), event.nilai(), berjasa);
    }

    /** Tahap agregasi: ringkasan terurut dari seluruh kejadian. */
    public static RingkasanAgregasi agregasi(List<KejadianTerproses> kejadian) {
        Map<String, BigDecimal> perKota = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kota,
                        TreeMap::new,
                        Collectors.reducing(BigDecimal.ZERO,
                                KejadianTerproses::nilaiBerjasa, BigDecimal::add)));
        Map<String, Long> perKategori = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kategori,
                        TreeMap::new, Collectors.counting()));
        BigDecimal totalNilai = kejadian.stream()
                .map(KejadianTerproses::nilai)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        BigDecimal totalBerjasa = kejadian.stream()
                .map(KejadianTerproses::nilaiBerjasa)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        return new RingkasanAgregasi(kejadian.size(), totalNilai, totalBerjasa,
                perKota, perKategori);
    }

    /** Tahap tulis: susun laporan akhir dari ringkasan dan hasil per baris. */
    public static List<String> tulis(RingkasanAgregasi ringkasan, List<HasilBaris> hasil) {
        int sah = 0;
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.Sah) {
                sah++;
            }
        }
        List<String> laporan = new ArrayList<>();
        laporan.add("=== Laporan pipeline penjualan ===");
        laporan.add("baris dibaca  : " + hasil.size());
        laporan.add("baris sah     : " + sah);
        laporan.add("baris ditolak : " + (hasil.size() - sah));
        laporan.add(String.format(Locale.US, "total nilai   : %,14.2f", ringkasan.totalNilai()));
        laporan.add(String.format(Locale.US, "total berjasa : %,14.2f", ringkasan.totalBerjasa()));
        laporan.add("nilai berjasa per kota:");
        ringkasan.nilaiBerjasaPerKota().forEach((kota, nilai) ->
                laporan.add(String.format(Locale.US, "  %-12s %,14.2f", kota, nilai)));
        laporan.add("jumlah transaksi per kategori:");
        ringkasan.jumlahPerKategori().forEach((kategori, jumlah) ->
                laporan.add(String.format(Locale.US, "  %-12s %6d", kategori, jumlah)));
        laporan.add("baris ditolak (terurut nomor):");
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.TidakSah(int nomor, String alasan)) {
                laporan.add(String.format(Locale.US, "  baris %2d: %s", nomor, alasan));
            }
        }
        return laporan;
    }

    /** Menjalankan seluruh tahapan secara sekuensial. */
    public static HasilPipeline jalankanSekuensial(List<String> isiBerkas) {
        List<BarisData> baris = baca(isiBerkas);
        List<HasilBaris> hasil = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            hasil.add(prosesSatuBaris(b));
        }
        return rakitHasil(baris.size(), hasil);
    }

    /**
     * Menjalankan tahap validasi + transformasi secara paralel: satu task per
     * baris di executor apa pun. Hasil dikumpulkan dulu, digabung, baru diurutkan.
     */
    public static HasilPipeline jalankanParalel(List<String> isiBerkas, ExecutorService eksekutor) {
        List<BarisData> baris = baca(isiBerkas);
        // Peta konkuren berkunci nomor baris: thread mana pun boleh menulis
        // bersamaan karena kunci tiap baris unik.
        Map<Integer, HasilBaris> hasilPerNomor = new ConcurrentHashMap<>();
        List<CompletableFuture<Void>> tugas = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            tugas.add(CompletableFuture.runAsync(() ->
                    hasilPerNomor.put(b.nomor(), prosesSatuBaris(b)), eksekutor));
        }
        // Gerbang tunggu tunggal: seluruh task selesai sebelum hasil dibaca.
        CompletableFuture.allOf(tugas.toArray(new CompletableFuture<?>[0])).join();
        // Pulihkan urutan nomor baris lewat salinan TreeMap.
        List<HasilBaris> terurut = new ArrayList<>(new TreeMap<>(hasilPerNomor).values());
        return rakitHasil(baris.size(), terurut);
    }

    /** Merakit hasil per baris menjadi ringkasan, laporan, dan statistik pipeline. */
    private static HasilPipeline rakitHasil(int jumlahBaris, List<HasilBaris> hasil) {
        List<KejadianTerproses> kejadian = new ArrayList<>(hasil.size());
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.Sah(var k)) {
                kejadian.add(k);
            }
        }
        RingkasanAgregasi ringkasan = agregasi(kejadian);
        return new HasilPipeline(jumlahBaris, kejadian.size(),
                jumlahBaris - kejadian.size(),
                ringkasan.nilaiBerjasaPerKota().size(),
                ringkasan.jumlahPerKategori().size(),
                tulis(ringkasan, hasil));
    }
}

Empat keputusan penting pada jalankanParalel:

  1. ConcurrentHashMap berkunci nomor baris (hasilPerNomor): thread mana pun boleh menulis bersamaan karena ConcurrentHashMap menjaga tiap entrinya dengan penguncian halus, dan kunci tiap baris unik sehingga tidak ada dua thread yang merebut entri yang sama.
  2. CompletableFuture.runAsync(..., eksekutor): task dijadwalkan pada executor yang kita pilih — pool tetap atau virtual threads — tanpa mengubah isi task.
  3. allOf(...).join() sebagai gerbang tunggu: join() menahan main sampai seluruh future selesai; pembacaan hasil hanya terjadi setelah titik ini.
  4. new TreeMap<>(hasilPerNomor).values(): salinan TreeMap mengiterasi kunci menaik, jadi hasil tersusun kembali menurut nomor baris meskipun thread selesai dalam urutan acak. rakitHasil dipakai bersama jalur sekuensial dan paralel, menjamin kedua jalur berbagi logika penyusunan yang sama.

Pool Tetap versus Virtual Threads #

Demo berikut menjalankan pipeline yang sama tiga kali — sekuensial, paralel pada pool tetap 4 thread, paralel pada virtual threads — lalu membandingkan ketiganya lewat equals() record:

java DemoPipeline.java 
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** Bagian 2: paralel vs sekuensial — hasil harus identik, urutan tetap terjaga. */
public class DemoPipeline {

    public static void main(String[] args) {
        List<String> isi = DataContoh.isiBerkas();
        HasilPipeline sekuensial = Pipeline.jalankanSekuensial(isi);

        HasilPipeline paralelPool;
        try (ExecutorService eksekutor = Executors.newFixedThreadPool(4)) {
            paralelPool = Pipeline.jalankanParalel(isi, eksekutor);
        }

        HasilPipeline paralelVirtual;
        try (ExecutorService eksekutor = Executors.newVirtualThreadPerTaskExecutor()) {
            paralelVirtual = Pipeline.jalankanParalel(isi, eksekutor);
        }

        System.out.println("== Perbandingan eksekusi pipeline (" + isi.size() + " baris) ==");
        ringkas("sekuensial             ", sekuensial);
        ringkas("paralel pool tetap 4   ", paralelPool);
        ringkas("paralel virtual thread ", paralelVirtual);
        System.out.println();
        System.out.println("hasil identik (sekuensial vs pool tetap) : "
                + sekuensial.equals(paralelPool));
        System.out.println("hasil identik (sekuensial vs virtual)    : "
                + sekuensial.equals(paralelVirtual));
        System.out.println("urutan laporan sama baris demi baris     : "
                + sekuensial.laporan().equals(paralelVirtual.laporan()));
    }

    private static void ringkas(String label, HasilPipeline hasil) {
        System.out.println(label + ": sah " + hasil.jumlahSah() + ", ditolak "
                + hasil.jumlahDitolak() + ", laporan " + hasil.laporan().size() + " baris");
    }
}

Output yang Dihasilkan:

text 
== Perbandingan eksekusi pipeline (30 baris) ==
sekuensial             : sah 26, ditolak 4, laporan 22 baris
paralel pool tetap 4   : sah 26, ditolak 4, laporan 22 baris
paralel virtual thread : sah 26, ditolak 4, laporan 22 baris

hasil identik (sekuensial vs pool tetap) : true
hasil identik (sekuensial vs virtual)    : true
urutan laporan sama baris demi baris     : true

Tiga pengamatan dari keluaran tersebut:

  1. Ketiga jalur menghasilkan statistik sama: 26 sah, 4 ditolak, 22 baris laporan. equals() pada record membandingkan seluruh field, dan dua perbandingan sama-sama true.
  2. Urutan laporan identik baris demi baris meskipun virtual thread selesai dalam urutan yang tidak menentu: TreeMap memulihkan urutan nomor baris sebelum tahap tulis berjalan.
  3. try-with-resources pada ExecutorService: sejak Java 19, close() menunggu seluruh task tuntas — gerbang tunggu paling ringkas sekaligus memastikan executor tidak memboroskan thread setelah blok selesai.

Antrean sebagai Titik Serah Terima #

Peta konkuren cocok saat hasil berakhir pada satu titik kumpul. Pola lain — tahap yang bekerja bergiliran seperti ban berjalan — memakai BlockingQueue: produsen put() (menunggu bila penuh), konsumen take() (menunggu bila kosong), dan kapasitas antrean menjadi katup tekanan balik alami:

java SerahTerimaAntrean.java 
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/** Pola produsen-konsumen: antrean berkapasitas sebagai titik serah terima. */
public class SerahTerimaAntrean {

    public static void main(String[] args) throws Exception {
        // Kapasitas 4: produsen menunggu bila antrean penuh (backpressure).
        BlockingQueue<String> antrean = new LinkedBlockingQueue<>(4);
        List<String> paketA = List.of("A1", "A2", "A3", "A4", "A5");
        List<String> paketB = List.of("B1", "B2", "B3", "B4", "B5");
        List<String> paketC = List.of("C1", "C2", "C3", "C4", "C5");

        List<String> terserap = new ArrayList<>(15);
        try (ExecutorService eksekutor = Executors.newVirtualThreadPerTaskExecutor()) {
            // Tiga produsen: satu virtual thread per produsen.
            eksekutor.submit(() -> taruhSemua(antrean, paketA));
            eksekutor.submit(() -> taruhSemua(antrean, paketB));
            eksekutor.submit(() -> taruhSemua(antrean, paketC));
            // Konsumen di main: mengambil tepat 15 paket yang dijanjikan.
            for (int i = 0; i < paketA.size() + paketB.size() + paketC.size(); i++) {
                terserap.add(antrean.take());
            }
        } // close(): menunggu seluruh produsen tuntas sebelum laporan dibuat

        System.out.println("== Antrean serah terima (kapasitas 4) ==");
        System.out.println("produsen        : 3 (virtual thread per produsen)");
        System.out.println("paket diserap   : " + terserap.size());
        System.out.println("sisa di antrean : " + antrean.size());
        terserap.sort(Comparator.naturalOrder());
        System.out.println("urutan serapan (terurut): " + terserap);
    }

    private static void taruhSemua(BlockingQueue<String> antrean, List<String> paket) {
        try {
            for (String p : paket) {
                antrean.put(p); // menunggu bila antrean penuh
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // pulihkan status interupsi
        }
    }
}

Output yang Dihasilkan:

text 
== Antrean serah terima (kapasitas 4) ==
produsen        : 3 (virtual thread per produsen)
paket diserap   : 15
sisa di antrean : 0
urutan serapan (terurut): [A1, A2, A3, A4, A5, B1, B2, B3, B4, B5, C1, C2, C3, C4, C5]

Tiga catatan dari contoh tersebut:

  1. Kapasitas 4 dengan 15 paket: produsen pasti pernah menunggu karena antrean hanya menampung 4 — tekanan balik ini menjaga memori tetap kecil saat tahap hilir lebih lambat dari hulu.
  2. Konsumen mengambil tepat 15 paket: jumlah diketahui di muka (3 produsen × 5 paket), jadi tidak perlu penanda akhir. Bila jumlah tak diketahui, kirim objek penanda khusus setelah produsen terakhir selesai.
  3. Cetak terjadi setelah close(): seluruh produsen dipastikan tuntas, daftar disortir, barulah laporan dicetak — aturan yang sama dengan jalankanParalel.

Perbandingan ringkas penampungan konkuren yang dipakai proyek ini:

StrukturPeran
ConcurrentHashMaphasil per baris berkunci unik
LinkedBlockingQueueserah terima antar tahap berkapasitas
ConcurrentLinkedQueuejejak kejadian dari banyak pekerja (Bagian 3)
LongAdderpenghitung yang sering ditambah bersamaan (Bagian 3 dan 4)

Pipeline kita kini cepat dan tetap tertib, tetapi masih rapuh: satu exception di tengah task meruntuhkan join() seluruh batch. Bagian 3 memberi pipeline kemampuan bertahan — kebijakan percobaan ulang, klasifikasi galat, dan laporan kegagalan yang tetap terurut.

Referensi #

Berkas Sumber

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

BarisData.java

Unduh berkas
java 
/** Satu baris berkas mentah: nomor urut (mulai dari 1) dan teks aslinya. */
public record BarisData(int nomor, String teks) {}

DataContoh.java

Unduh berkas
java 
import java.util.List;

/**
 * Isi berkas penjualan contoh: 30 baris data kuartal pertama 2026.
 * Dua puluh enam baris sah dan empat baris sengaja cacat untuk menguji validator.
 */
public final class DataContoh {

    private DataContoh() {
        // Kelas utilitas: seluruh anggota static.
    }

    /** Isi berkas contoh; pada aplikasi nyata berasal dari Files.readAllLines. */
    public static List<String> isiBerkas() {
        return List.of(
                "2026-01-05;Bandung;Budi;Elektronik;1250000",
                "2026-01-07;Surabaya;Siti;Pakaian;340000",
                "2026-01-09;bandung;Dewi;Buku;95000",
                "2026-01-12;Yogyakarta;Budi;Peralatan;780000",
                "2026-01-15;Medan;Siti;Elektronik;2100000",
                "2026-01-18;Bandung;Dewi;Buku;128000",
                "2026-01-20;Medan;Siti;Pakaian;-150000",
                "2026-01-21;Makassar;Budi;Pakaian;450000",
                "2026-01-24;surabaya;Siti;Peralatan;620000",
                "2026-01-27;Yogyakarta;Dewi;Elektronik;1850000",
                "2026-02-02;Bandung;Siti;Pakaian;295000",
                "2026-02-05;Bandung;Dewi;Buku",
                "2026-02-08;Surabaya;Budi;Elektronik;990000",
                "2026-02-11;Makassar;Siti;Camilan;88000",
                "2026-02-14;Yogyakarta;Dewi;Pakaian;380000",
                "2026-02-17;Medan;Budi;Elektronik;1720000",
                "2026-02-20;bandung;Siti;Buku;112000",
                "2026-02-23;Surabaya;Dewi;Peralatan;690000",
                "2026-02-26;Makassar;Budi;Pakaian;410000",
                "2026-02-28;Yogyakarta;Siti;Elektronik;1580000",
                "2026-03-04;Medan;Dewi;Buku;64000",
                "2026-03-07;Bandung;Budi;Peralatan;830000",
                "2026-03-10;Surabaya;Siti;Elektronik;1350000",
                "2026-03-13;Makassar;Dewi;Buku;88000",
                "2026-03-16;Yogyakarta;Budi;Pakaian;365000",
                "2026-03-19;Bandung;Siti;Peralatan;575000",
                "2026-03-22;medan;Dewi;Elektronik;1980000",
                "2026-03-25;Surabaya;Budi;Buku;102000",
                "2026-03-28;Makassar;Siti;Pakaian;430000",
                "2026-13-31;Yogyakarta;Dewi;Peralatan;715000");
    }
}

DemoPipeline.java

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

/** Bagian 2: paralel vs sekuensial — hasil harus identik, urutan tetap terjaga. */
public class DemoPipeline {

    public static void main(String[] args) {
        List<String> isi = DataContoh.isiBerkas();
        HasilPipeline sekuensial = Pipeline.jalankanSekuensial(isi);

        HasilPipeline paralelPool;
        try (ExecutorService eksekutor = Executors.newFixedThreadPool(4)) {
            paralelPool = Pipeline.jalankanParalel(isi, eksekutor);
        }

        HasilPipeline paralelVirtual;
        try (ExecutorService eksekutor = Executors.newVirtualThreadPerTaskExecutor()) {
            paralelVirtual = Pipeline.jalankanParalel(isi, eksekutor);
        }

        System.out.println("== Perbandingan eksekusi pipeline (" + isi.size() + " baris) ==");
        ringkas("sekuensial             ", sekuensial);
        ringkas("paralel pool tetap 4   ", paralelPool);
        ringkas("paralel virtual thread ", paralelVirtual);
        System.out.println();
        System.out.println("hasil identik (sekuensial vs pool tetap) : "
                + sekuensial.equals(paralelPool));
        System.out.println("hasil identik (sekuensial vs virtual)    : "
                + sekuensial.equals(paralelVirtual));
        System.out.println("urutan laporan sama baris demi baris     : "
                + sekuensial.laporan().equals(paralelVirtual.laporan()));
    }

    private static void ringkas(String label, HasilPipeline hasil) {
        System.out.println(label + ": sah " + hasil.jumlahSah() + ", ditolak "
                + hasil.jumlahDitolak() + ", laporan " + hasil.laporan().size() + " baris");
    }
}

EventPenjualan.java

Unduh berkas
java 
import java.math.BigDecimal;
import java.time.LocalDate;

/**
 * Kontrak keluaran tahap validasi: event penjualan yang lolos seluruh pemeriksaan.
 * Tahap transformasi menerima record ini tanpa memeriksa format lagi.
 */
public record EventPenjualan(int nomorBaris, LocalDate tanggal, String kota,
                             String pelanggan, String kategori, BigDecimal nilai) {}

HasilBaris.java

Unduh berkas
java 
/**
 * Kontrak hasil pemrosesan satu baris: nilai data, bukan exception.
 * Varian terkunci lewat sealed sehingga compiler memaksa setiap switch
 * menangani semua kemungkinan.
 */
public sealed interface HasilBaris permits HasilBaris.Sah, HasilBaris.TidakSah {

    /** Nomor baris sumber — dipakai pelaporan galat dan pengurutan hasil. */
    int nomorBaris();

    /** Baris selesai diproses dan membawa kejadian siap agregasi. */
    record Sah(KejadianTerproses kejadian) implements HasilBaris {
        @Override
        public int nomorBaris() {
            return kejadian.nomorBaris();
        }
    }

    /** Baris ditolak validator; alasan ditulis untuk laporan galat. */
    record TidakSah(int nomorBaris, String alasan) implements HasilBaris {
        @Override
        public int nomorBaris() {
            return nomorBaris;
        }
    }
}

HasilPipeline.java

Unduh berkas
java 
import java.util.List;

/** Hasil pipeline utuh: statistik tiap tahap plus laporan akhir siap cetak. */
public record HasilPipeline(int jumlahBaris,
                            int jumlahSah,
                            int jumlahDitolak,
                            int jumlahKota,
                            int jumlahKategori,
                            List<String> laporan) {}

KejadianTerproses.java

Unduh berkas
java 
import java.math.BigDecimal;
import java.time.LocalDate;

/**
 * Kontrak keluaran tahap transformasi: data siap agregasi.
 * Kota sudah dibakukan (huruf kapital) dan nilai berjasa sudah dihitung.
 */
public record KejadianTerproses(int nomorBaris, LocalDate tanggal, String kota,
                                String pelanggan, String kategori, BigDecimal nilai,
                                BigDecimal nilaiBerjasa) {}

Pipeline.java

Unduh berkas
java 
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/** Pipeline penjualan: baca -> validasi -> transformasi -> agregasi -> tulis. */
public final class Pipeline {

    /** Kategori yang dikenal validator; di luar daftar ini baris ditolak. */
    private static final Set<String> KATEGORI_DIKENAL =
            Set.of("Elektronik", "Pakaian", "Buku", "Peralatan");

    /** Faktor biaya layanan 2% yang dihitung tahap transformasi. */
    private static final BigDecimal BIAYA_LAYANAN = new BigDecimal("1.02");

    private Pipeline() {
        // Kelas utilitas: seluruh anggota static.
    }

    /** Tahap baca: memberi nomor urut pada tiap baris berkas. */
    public static List<BarisData> baca(List<String> isiBerkas) {
        List<BarisData> baris = new ArrayList<>(isiBerkas.size());
        for (int i = 0; i < isiBerkas.size(); i++) {
            baris.add(new BarisData(i + 1, isiBerkas.get(i)));
        }
        return baris;
    }

    /** Tahap validasi lalu transformasi untuk satu baris. */
    public static HasilBaris prosesSatuBaris(BarisData baris) {
        String[] segmen = baris.teks().split(";", -1);
        if (segmen.length != 5) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "jumlah segmen " + segmen.length + ", harus 5");
        }
        LocalDate tanggal;
        try {
            tanggal = LocalDate.parse(segmen[0]);
        } catch (DateTimeParseException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "tanggal tidak terurai: '" + segmen[0] + "'");
        }
        String kota = segmen[1].trim();
        String pelanggan = segmen[2].trim();
        String kategori = segmen[3].trim();
        if (kota.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "kota kosong");
        }
        if (pelanggan.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "nama pelanggan kosong");
        }
        if (!KATEGORI_DIKENAL.contains(kategori)) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "kategori tidak dikenal: '" + kategori + "'");
        }
        BigDecimal nilai;
        try {
            nilai = new BigDecimal(segmen[4].trim());
        } catch (NumberFormatException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai tidak terurai: '" + segmen[4].trim() + "'");
        }
        if (nilai.signum() <= 0) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai harus positif: " + nilai.toPlainString());
        }
        EventPenjualan event = new EventPenjualan(baris.nomor(), tanggal, kota,
                pelanggan, kategori, nilai);
        return new HasilBaris.Sah(ubah(event));
    }

    /** Tahap transformasi: bakukan kota dan hitung nilai berjasa 2%. */
    public static KejadianTerproses ubah(EventPenjualan event) {
        BigDecimal berjasa = event.nilai().multiply(BIAYA_LAYANAN)
                .setScale(2, RoundingMode.HALF_EVEN);
        return new KejadianTerproses(event.nomorBaris(), event.tanggal(),
                event.kota().toUpperCase(Locale.ROOT), event.pelanggan(),
                event.kategori(), event.nilai(), berjasa);
    }

    /** Tahap agregasi: ringkasan terurut dari seluruh kejadian. */
    public static RingkasanAgregasi agregasi(List<KejadianTerproses> kejadian) {
        Map<String, BigDecimal> perKota = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kota,
                        TreeMap::new,
                        Collectors.reducing(BigDecimal.ZERO,
                                KejadianTerproses::nilaiBerjasa, BigDecimal::add)));
        Map<String, Long> perKategori = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kategori,
                        TreeMap::new, Collectors.counting()));
        BigDecimal totalNilai = kejadian.stream()
                .map(KejadianTerproses::nilai)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        BigDecimal totalBerjasa = kejadian.stream()
                .map(KejadianTerproses::nilaiBerjasa)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        return new RingkasanAgregasi(kejadian.size(), totalNilai, totalBerjasa,
                perKota, perKategori);
    }

    /** Tahap tulis: susun laporan akhir dari ringkasan dan hasil per baris. */
    public static List<String> tulis(RingkasanAgregasi ringkasan, List<HasilBaris> hasil) {
        int sah = 0;
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.Sah) {
                sah++;
            }
        }
        List<String> laporan = new ArrayList<>();
        laporan.add("=== Laporan pipeline penjualan ===");
        laporan.add("baris dibaca  : " + hasil.size());
        laporan.add("baris sah     : " + sah);
        laporan.add("baris ditolak : " + (hasil.size() - sah));
        laporan.add(String.format(Locale.US, "total nilai   : %,14.2f", ringkasan.totalNilai()));
        laporan.add(String.format(Locale.US, "total berjasa : %,14.2f", ringkasan.totalBerjasa()));
        laporan.add("nilai berjasa per kota:");
        ringkasan.nilaiBerjasaPerKota().forEach((kota, nilai) ->
                laporan.add(String.format(Locale.US, "  %-12s %,14.2f", kota, nilai)));
        laporan.add("jumlah transaksi per kategori:");
        ringkasan.jumlahPerKategori().forEach((kategori, jumlah) ->
                laporan.add(String.format(Locale.US, "  %-12s %6d", kategori, jumlah)));
        laporan.add("baris ditolak (terurut nomor):");
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.TidakSah(int nomor, String alasan)) {
                laporan.add(String.format(Locale.US, "  baris %2d: %s", nomor, alasan));
            }
        }
        return laporan;
    }

    /** Menjalankan seluruh tahapan secara sekuensial. */
    public static HasilPipeline jalankanSekuensial(List<String> isiBerkas) {
        List<BarisData> baris = baca(isiBerkas);
        List<HasilBaris> hasil = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            hasil.add(prosesSatuBaris(b));
        }
        return rakitHasil(baris.size(), hasil);
    }

    /**
     * Menjalankan tahap validasi + transformasi secara paralel: satu task per
     * baris di executor apa pun. Hasil dikumpulkan dulu, digabung, baru diurutkan.
     */
    public static HasilPipeline jalankanParalel(List<String> isiBerkas, ExecutorService eksekutor) {
        List<BarisData> baris = baca(isiBerkas);
        // Peta konkuren berkunci nomor baris: thread mana pun boleh menulis
        // bersamaan karena kunci tiap baris unik.
        Map<Integer, HasilBaris> hasilPerNomor = new ConcurrentHashMap<>();
        List<CompletableFuture<Void>> tugas = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            tugas.add(CompletableFuture.runAsync(() ->
                    hasilPerNomor.put(b.nomor(), prosesSatuBaris(b)), eksekutor));
        }
        // Gerbang tunggu tunggal: seluruh task selesai sebelum hasil dibaca.
        CompletableFuture.allOf(tugas.toArray(new CompletableFuture<?>[0])).join();
        // Pulihkan urutan nomor baris lewat salinan TreeMap.
        List<HasilBaris> terurut = new ArrayList<>(new TreeMap<>(hasilPerNomor).values());
        return rakitHasil(baris.size(), terurut);
    }

    /** Merakit hasil per baris menjadi ringkasan, laporan, dan statistik pipeline. */
    private static HasilPipeline rakitHasil(int jumlahBaris, List<HasilBaris> hasil) {
        List<KejadianTerproses> kejadian = new ArrayList<>(hasil.size());
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.Sah(var k)) {
                kejadian.add(k);
            }
        }
        RingkasanAgregasi ringkasan = agregasi(kejadian);
        return new HasilPipeline(jumlahBaris, kejadian.size(),
                jumlahBaris - kejadian.size(),
                ringkasan.nilaiBerjasaPerKota().size(),
                ringkasan.jumlahPerKategori().size(),
                tulis(ringkasan, hasil));
    }
}

RingkasanAgregasi.java

Unduh berkas
java 
import java.math.BigDecimal;
import java.util.Map;

/**
 * Kontrak keluaran tahap agregasi: ringkasan seluruh kejadian.
 * Peta memakai TreeMap sehingga iterasi selalu terurut (deterministik).
 */
public record RingkasanAgregasi(int jumlahKejadian,
                                BigDecimal totalNilai,
                                BigDecimal totalBerjasa,
                                Map<String, BigDecimal> nilaiBerjasaPerKota,
                                Map<String, Long> jumlahPerKategori) {}

SerahTerimaAntrean.java

Unduh berkas
java 
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/** Pola produsen-konsumen: antrean berkapasitas sebagai titik serah terima. */
public class SerahTerimaAntrean {

    public static void main(String[] args) throws Exception {
        // Kapasitas 4: produsen menunggu bila antrean penuh (backpressure).
        BlockingQueue<String> antrean = new LinkedBlockingQueue<>(4);
        List<String> paketA = List.of("A1", "A2", "A3", "A4", "A5");
        List<String> paketB = List.of("B1", "B2", "B3", "B4", "B5");
        List<String> paketC = List.of("C1", "C2", "C3", "C4", "C5");

        List<String> terserap = new ArrayList<>(15);
        try (ExecutorService eksekutor = Executors.newVirtualThreadPerTaskExecutor()) {
            // Tiga produsen: satu virtual thread per produsen.
            eksekutor.submit(() -> taruhSemua(antrean, paketA));
            eksekutor.submit(() -> taruhSemua(antrean, paketB));
            eksekutor.submit(() -> taruhSemua(antrean, paketC));
            // Konsumen di main: mengambil tepat 15 paket yang dijanjikan.
            for (int i = 0; i < paketA.size() + paketB.size() + paketC.size(); i++) {
                terserap.add(antrean.take());
            }
        } // close(): menunggu seluruh produsen tuntas sebelum laporan dibuat

        System.out.println("== Antrean serah terima (kapasitas 4) ==");
        System.out.println("produsen        : 3 (virtual thread per produsen)");
        System.out.println("paket diserap   : " + terserap.size());
        System.out.println("sisa di antrean : " + antrean.size());
        terserap.sort(Comparator.naturalOrder());
        System.out.println("urutan serapan (terurut): " + terserap);
    }

    private static void taruhSemua(BlockingQueue<String> antrean, List<String> paket) {
        try {
            for (String p : paket) {
                antrean.put(p); // menunggu bila antrean penuh
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // pulihkan status interupsi
        }
    }
}