Thread Pools

ExecutorService membebaskan kita dari pembuatan thread manual, tetapi pabrik Executors menawarkan banyak varian — dan pilihan yang salah sama sekali tidak terasa pada beban uji kecil, lalu muncul sebagai kehabisan memori atau kelaparan antrean saat produksi. Pool yang terlalu kecil membuat task menginap di antrean; pool yang terlalu besar membuat thread saling berebut CPU yang sama plus biaya konteks switch yang tidak perlu. Bab ini membandingkan empat jenis pool secara langsung dalam satu program, menyuji penjadwalan berbasis waktu, lalu menutup dengan cara memperkirakan ukuran pool yang tepat untuk dua kelas beban: CPU-bound dan I/O-bound.

Empat Jenis Pool dalam Satu Program #

Pabrik Executors menyediakan empat bentuk utama. newFixedThreadPool(n) memelihara tepat n thread dengan satu antrean tanpa batas. newCachedThreadPool() membuat thread sesuai permintaan dan memensiunkan thread yang menganggur 60 detik. newWorkStealingPool(n) membangun ForkJoinPool yang tiap pekerjanya memiliki antrean sendiri dan saling mencuri pekerjaan. newVirtualThreadPerTaskExecutor() — bentuk termuda — menjalankan setiap task pada virtual thread segar:

java
JenisPool.java
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class JenisPool {
    // Jalankan 6 task yang sama pada satu eksekutor, kumpulkan hasilnya
    // lewat Future.get() berurutan -- hasil selalu terurut sesuai pengiriman.
    private static List<Integer> jalankan(ExecutorService eksekutor) throws Exception {
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 1; i <= 6; i++) {
            int nomor = i;
            futures.add(eksekutor.submit(() -> {
                Thread.sleep(20); // meniru pekerjaan singkat
                return nomor * nomor; // kuadrat nomor task
            }));
        }
        List<Integer> hasil = new ArrayList<>();
        for (Future<Integer> f : futures) {
            hasil.add(f.get());
        }
        eksekutor.shutdown();
        return hasil;
    }

    public static void main(String[] args) throws Exception {
        // Empat jenis eksekutor, satu pekerjaan yang sama.
        System.out.println("fixed 2 thread        : "
                + jalankan(Executors.newFixedThreadPool(2)));
        System.out.println("cached                : "
                + jalankan(Executors.newCachedThreadPool()));
        System.out.println("work-stealing 2 pekerja: "
                + jalankan(Executors.newWorkStealingPool(2)));
        System.out.println("virtual per task      : "
                + jalankan(Executors.newVirtualThreadPerTaskExecutor()));
    }
}

Output yang Dihasilkan:

text
fixed 2 thread        : [1, 4, 9, 16, 25, 36]
cached                : [1, 4, 9, 16, 25, 36]
work-stealing 2 pekerja: [1, 4, 9, 16, 25, 36]
virtual per task      : [1, 4, 9, 16, 25, 36]

Hasil keempat baris identik — dan memang itu intinya. Bedah singkatnya:

Enam task, hasil [1, 4, 9, 16, 25, 36] di semua pool: task ke-i mengembalikan i², dikumpulkan lewat get() sesuai urutan pengiriman. Perbedaan keempat pool terletak pada mekanisme eksekusi, bukan pada hasil — kode task tidak perlu tahu pool apa yang menjalankannya.
Fixed 2 thread: keenam task antre di dua thread pekerja; Thread.sleep(20) meniru pekerjaan sehingga total durasi sekitar tiga gelombang — pembagian rapi yang mudah diprediksi.
Cached: enam thread dibuat nyaris serentak karena semua task datang hampir bersamaan; pada beban sesungguhnya, pool ini menumbuhkan thread tanpa batas bila task datang lebih cepat daripada diselesaikan — perhatikan peringatan di bawah.
Virtual per task: enam virtual thread ringan, satu per task, tanpa antrean sama sekali — calon bawaan untuk beban I/O padat yang dibahas khusus pada Chapter Virtual Threads.

Menjadwalkan dengan ScheduledExecutorService #

Timer lawas memiliki cacat desain (satu thread saja, sensitif terhadap exception, dan jadwal bergeser). Penggantinya adalah ScheduledExecutorService: pool yang mampu menjalankan task setelah jeda tertentu:

java
JadwalTugas.java
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class JadwalTugas {
    public static void main(String[] args) throws Exception {
        ScheduledExecutorService penjadwal = Executors.newScheduledThreadPool(1);
        try {
            // Tiga tugas sekali jalan dengan jeda awal berbeda.
            // Urutan waktu jadwal: malam (100 ms) -> siang (200 ms) -> pagi (300 ms).
            ScheduledFuture<String> pagi =
                    penjadwal.schedule(() -> "alarm pagi", 300, TimeUnit.MILLISECONDS);
            ScheduledFuture<String> siang =
                    penjadwal.schedule(() -> "alarm siang", 200, TimeUnit.MILLISECONDS);
            ScheduledFuture<String> malam =
                    penjadwal.schedule(() -> "alarm malam", 100, TimeUnit.MILLISECONDS);

            // get() menunggu jadwal masing-masing tiba; hasil dicetak sesuai
            // urutan variabel, bukan urutan waktu jadwal -- tetap deterministik.
            System.out.println("jadwal ke-1 (jeda 300 ms): " + pagi.get());
            System.out.println("jadwal ke-2 (jeda 200 ms): " + siang.get());
            System.out.println("jadwal ke-3 (jeda 100 ms): " + malam.get());
            System.out.println("semua alarm sudah berbunyi: "
                    + (pagi.isDone() && siang.isDone() && malam.isDone()));
        } finally {
            penjadwal.shutdown();
            if (!penjadwal.awaitTermination(5, TimeUnit.SECONDS)) {
                penjadwal.shutdownNow();
            }
        }
    }
}

Output yang Dihasilkan:

text
jadwal ke-1 (jeda 300 ms): alarm pagi
jadwal ke-2 (jeda 200 ms): alarm siang
jadwal ke-3 (jeda 100 ms): alarm malam
semua alarm sudah berbunyi: true

Tiga catatan dari contoh jadwal tersebut:

schedule(task, jeda, satuan): task dijalankan sekali setelah jeda berlalu. Nilai baliknya ScheduledFuture, turunan Future yang juga bisa ditanyai sisa jeda lewat getDelay().
Urutan cetak tidak mengikuti urutan jadwal: alarm malam berbunyi paling awal (jeda 100 milidetik), tetapi barisnya tercetak terakhir karena get() dipanggil pada urutan itu. Pemisahan “urutan selesai” dan “urutan cetak” adalah pola deterministik yang sama dengan bab-bab sebelumnya.
Pool beranggota satu: penjadwal satu thread cukup untuk tugas ringan; tambah pekerja bila tugas terjadwal bisa berjalan paralel.

Untuk jadwal berulang, dua varian tersedia — scheduleAtFixedRate (periode tetap antar awal eksekusi) dan scheduleAtFixedDelay (jeda tetap antar akhir dan awal berikutnya) — keduanya dihentikan lewat cancel() pada ScheduledFuture-nya:

java
ScheduledFuture<?> pembersih = penjadwal.scheduleAtFixedRate(
        () -> bersihkanCache(),   // tugas berulang
        0, 5, TimeUnit.MINUTES);  // mulai sekarang, tiap 5 menit
// ... di titik shutdown aplikasi:
pembersih.cancel(false);          // false: jangan interupsi yang sedang berjalan

Berapa Ukuran Pool yang Tepat #

Untuk beban CPU-bound — komputasi murni tanpa menunggu apa pun — tambahan thread di atas jumlah core tidak menambah throughput, hanya menambah konteks switch. Patokannya sederhana: N_threads ≈ N_cpu. Untuk beban I/O-bound, thread menganggur besar-besaran selama menunggu jaringan atau disk, sehingga jumlah thread boleh jauh lebih banyak. Rumus klasik dari Brian Goetz merangkum keduanya:

text
N_threads = N_cpu * U_cpu * (1 + W/C)

N_cpu : jumlah core prosesor
U_cpu : target pemanfaatan CPU (0 sampai 1)
W/C   : rasio waktu tunggu (wait) terhadap waktu komputasi (compute)

Contoh konkret pada mesin 2 core: beban komputasi murni (W/C = 0) memberi 2 × 1 × (1 + 0) = 2 thread; beban layanan web yang 90% waktunya menunggu respons basis data (W/C = 9) memberi 2 × 1 × (1 + 9) = 20 thread. Angka hasil rumus adalah titik mulai pengukuran, bukan dogma — profil dan ukur beban nyata sebelum memfinalkan.

Memilih Pool #

Situasi	Pool yang tepat	Alasan
Beban CPU-bound stabil	`newFixedThreadPool(N_cpu)`	Paralelisme maksimal tanpa pemborosan
Beban I/O-bound, task menunggu lama	`newVirtualThreadPerTaskExecutor()`	Jutaan penunggu ringan; thread mahal tidak terpakai menunggu
Task kecil bervolume tak terduga	`newFixedThreadPool` dengan antrean dibatasi (diimplementasi manual)	Elastis tanpa risiko tanpa batas
Komputasi rekursif divide-and-conquer	`newWorkStealingPool` / `ForkJoinPool`	Work stealing menyeimbangkan beban; dibahas pada Chapter Fork/Join Framework
Tugas berbasis waktu	`newScheduledThreadPool`	Jeda, periode, dan pembatalan terjadwal
Prototipe / uji cepat	`newCachedThreadPool()`	Tanpa konfigurasi; hindari di produksi

Satu kebutuhan belum terjawab oleh semua pool di atas: hasil task sering perlu diolah segera setelah tersedia, tanpa memblokir pemanggil di get(). Jawaban modernnya — CompletableFuture dengan rantai tahapan — menjadi topik Chapter berikutnya.

Referensi #

Berkas Sumber

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

java
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class JadwalTugas {
    public static void main(String[] args) throws Exception {
        ScheduledExecutorService penjadwal = Executors.newScheduledThreadPool(1);
        try {
            // Tiga tugas sekali jalan dengan jeda awal berbeda.
            // Urutan waktu jadwal: malam (100 ms) -> siang (200 ms) -> pagi (300 ms).
            ScheduledFuture<String> pagi =
                    penjadwal.schedule(() -> "alarm pagi", 300, TimeUnit.MILLISECONDS);
            ScheduledFuture<String> siang =
                    penjadwal.schedule(() -> "alarm siang", 200, TimeUnit.MILLISECONDS);
            ScheduledFuture<String> malam =
                    penjadwal.schedule(() -> "alarm malam", 100, TimeUnit.MILLISECONDS);

            // get() menunggu jadwal masing-masing tiba; hasil dicetak sesuai
            // urutan variabel, bukan urutan waktu jadwal -- tetap deterministik.
            System.out.println("jadwal ke-1 (jeda 300 ms): " + pagi.get());
            System.out.println("jadwal ke-2 (jeda 200 ms): " + siang.get());
            System.out.println("jadwal ke-3 (jeda 100 ms): " + malam.get());
            System.out.println("semua alarm sudah berbunyi: "
                    + (pagi.isDone() && siang.isDone() && malam.isDone()));
        } finally {
            penjadwal.shutdown();
            if (!penjadwal.awaitTermination(5, TimeUnit.SECONDS)) {
                penjadwal.shutdownNow();
            }
        }
    }
}

java
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class JenisPool {
    // Jalankan 6 task yang sama pada satu eksekutor, kumpulkan hasilnya
    // lewat Future.get() berurutan -- hasil selalu terurut sesuai pengiriman.
    private static List<Integer> jalankan(ExecutorService eksekutor) throws Exception {
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 1; i <= 6; i++) {
            int nomor = i;
            futures.add(eksekutor.submit(() -> {
                Thread.sleep(20); // meniru pekerjaan singkat
                return nomor * nomor; // kuadrat nomor task
            }));
        }
        List<Integer> hasil = new ArrayList<>();
        for (Future<Integer> f : futures) {
            hasil.add(f.get());
        }
        eksekutor.shutdown();
        return hasil;
    }

    public static void main(String[] args) throws Exception {
        // Empat jenis eksekutor, satu pekerjaan yang sama.
        System.out.println("fixed 2 thread        : "
                + jalankan(Executors.newFixedThreadPool(2)));
        System.out.println("cached                : "
                + jalankan(Executors.newCachedThreadPool()));
        System.out.println("work-stealing 2 pekerja: "
                + jalankan(Executors.newWorkStealingPool(2)));
        System.out.println("virtual per task      : "
                + jalankan(Executors.newVirtualThreadPerTaskExecutor()));
    }
}