KAL
Menu

Collections Konkuren

Berbagi data antar-thread dengan ConcurrentHashMap, CopyOnWriteArrayList, dan blocking queue.

Daftar isi

Saat dua thread memakai satu koleksi biasa, hasilnya tidak dapat dipercaya. Pembaruan pada HashMap bisa hilang karena dua thread menulis alamat bucket yang sama, dan iterasi ArrayList yang dimodifikasi di tengah jalan memunculkan ConcurrentModificationException. Paket java.util.concurrent menyediakan koleksi yang dirancang untuk dipakai bersama dari awal. Topik ini membahas tiga yang paling sering dipakai, yaitu ConcurrentHashMap untuk peta bersama, CopyOnWriteArrayList untuk daftar yang jarang ditulis, dan BlockingQueue untuk serah terima tugas antar-thread.

ConcurrentHashMap untuk peta bersama#

ConcurrentHashMap menggantikan HashMap pada kode yang diakses banyak thread. Pengunciannya halus, yaitu thread yang menulis ke satu bucket tidak mengunci seluruh peta, sehingga pembacaan dan penulisan di bagian lain tetap berjalan. Yang lebih penting, kelas ini menyediakan operasi atomik seperti merge dan compute, yang melakukan baca-ubah-tulis dalam satu langkah terlindungi, persis persoalan yang sebelumnya ditangani synchronized secara manual.

java DemoPetaKonkuren.java 
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class DemoPetaKonkuren {

    static Map<String, Integer> hitungan = new ConcurrentHashMap<>();

    static void hitungKata(String[] bagian) {
        for (String kata : bagian) {
            hitungan.merge(kata, 1, Integer::sum);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        String[] bagianA = {"java", "thread", "java", "map"};
        String[] bagianB = {"java", "map", "java", "java"};

        Thread t1 = new Thread(() -> hitungKata(bagianA));
        Thread t2 = new Thread(() -> hitungKata(bagianB));

        t1.start();
        t2.start();
        t1.join();
        t2.join();

        System.out.println("java  : " + hitungan.get("java"));
        System.out.println("map   : " + hitungan.get("map"));
        System.out.println("thread: " + hitungan.get("thread"));
    }
}
text Keluaran 
java  : 5
map   : 2
thread: 1

Dua thread menghitung dua bagian daftar kata, dan kata java serta map disentuh keduanya. Pemanggilan merge(kata, 1, Integer::sum) menambah satu bila kata belum ada, atau menjumlahkan nilai lamanya bila sudah, dan seluruh langkah itu atomik sehingga tidak ada perhitungan yang hilang. Pada HashMap biasa dengan penambahan manual, sebagian kenaikan justru lenyap seperti yang disaksikan pada topik sinkronisasi. Satu batasan perlu diingat, kunci maupun nilai pada ConcurrentHashMap tidak boleh null.

CopyOnWriteArrayList untuk daftar yang jarang berubah#

CopyOnWriteArrayList memilih taktik yang berbeda, yaitu setiap penulisan membuat salinan array baru, mengubah salinannya, lalu menukarnya dengan array lama dalam satu operasi yang aman dibaca. Iterator yang sedang berjalan memegang array lama sebagai snapshot, yakni potret daftar pada saat iterator dibuat, sehingga daftar boleh diubah sekalipun sedang diiterasi. Biayanya jelas, menulis mahal karena menyalin seluruh isi, sehingga kelas ini pas untuk daftar yang dibaca sering dan diubah jarang, misalnya daftar pemantau atau daftar piket.

java DemoCOW.java 
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class DemoCOW {

    public static void main(String[] args) {
        List<String> daftarPiket = new CopyOnWriteArrayList<>();
        daftarPiket.add("Ali");
        daftarPiket.add("Budi");
        daftarPiket.add("Citra");

        for (String nama : daftarPiket) {
            System.out.println("Mengingatkan: " + nama);
            daftarPiket.remove(nama);
        }

        System.out.println("Sisa list   : " + daftarPiket);
    }
}
text Keluaran 
Mengingatkan: Ali
Mengingatkan: Budi
Mengingatkan: Citra
Sisa list   : []

Kode di atas menghapus elemen pada saat iterasi berjalan, sesuatu yang pada ArrayList biasa memicu ConcurrentModificationException. Iterator CopyOnWriteArrayList mengabaikan perubahan yang terjadi setelah ia dibuat, sehingga perulangannya menyelesaikan potret awalnya dengan tenang sampai tiga nama habis diingatkan. Perilaku snapshot ini juga berarti iterator tidak pernah melihat penambahan yang dilakukan thread lain di tengah jalan, dan hal itu perlu diantisipasi pada desainnya.

BlockingQueue untuk serah terima tugas#

BlockingQueue adalah antrean yang dibekali kemampuan menunggu. Metode put menunggu sampai antrean penuhnya punya ruang, sedangkan take menunggu sampai ada elemen yang bisa diambil, sehingga produsen dan konsumen saling mengatur kecepatannya tanpa wait dan notify manual. Kapasitas antrean yang dibatasi, misalnya ArrayBlockingQueue berkapasitas dua, membuat produsen yang terlalu cepat ikut melambat, dan itulah mekanisme penahan beban yang sehat.

java DemoAntrian.java 
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class DemoAntrian {

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> antrian = new ArrayBlockingQueue<>(2);

        Thread produsen = new Thread(() -> {
            try {
                for (int i = 1; i <= 3; i++) {
                    antrian.put("tiket-" + i);
                }
                antrian.put("SELESAI");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread konsumen = new Thread(() -> {
            try {
                while (true) {
                    String tugas = antrian.take();
                    if (tugas.equals("SELESAI")) {
                        break;
                    }
                    System.out.println("Diproses: " + tugas);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        produsen.start();
        konsumen.start();
        produsen.join();
        konsumen.join();
        System.out.println("Antrian kosong, program berakhir");
    }
}
text Keluaran 
Diproses: tiket-1
Diproses: tiket-2
Diproses: tiket-3
Antrian kosong, program berakhir

Produsen menaruh tiga tiket lalu satu sinyal khusus, dan sinyal berhenti semacam itu dikenal sebagai poison pill, yaitu nilai yang sengaja disisipkan agar konsumen tahu pekerjaannya selesai. Konsumen berputar mengambil elemen, memprosesnya, dan berhenti tepat ketika sinyalnya muncul. Karena antrean berperilaku masuk-pertama-keluar-pertama dan konsumennya satu, urutan keluarannya selalu rapi meski kedua thread bergantian berebut waktu prosesor.

Kesalahan umum#

  • Memakai HashMap atau ArrayList biasa untuk data lintas-thread. Pembaruan yang hilang dan exception saat iterasi adalah akibatnya. Pilih koleksi dari paket java.util.concurrent sesuai pola baca-tulisnya.
  • Mengira size() dan operasi kumpulan pada ConcurrentHashMap adalah satu potret. Nilainya konsisten per pemanggilan, tetapi menghitung lalu memutuskan berdasarkan hasilnya bukan langkah atomik. Gunakan merge, compute, atau putIfAbsent untuk keputusan yang harus utuh.
  • Memakai CopyOnWriteArrayList pada daftar yang sering ditulis. Setiap penulisan menyalin seluruh array, sehingga beban penulisan yang tinggi membuat program melambat sendiri. Kelas ini milik pola baca-sering.
  • Memakai antrean tanpa batas untuk tugas dari pihak luar. LinkedBlockingQueue tanpa kapasitas membuat antrean membengkak ketika konsumen kalah cepat. Batasi kapasitasnya agar tekanan terlihat dan produsen ikut menunggu.

Ringkasan#

  • ConcurrentHashMap aman dipakai bersama dan menyediakan operasi atomik seperti merge serta compute untuk baca-ubah-tulis, tanpa menerima null.
  • CopyOnWriteArrayList menyalin array pada setiap tulisan, dan iteratornya membaca snapshot sehingga aman sambil dimodifikasi.
  • BlockingQueue mengatur kecepatan produsen dan konsumen lewat put serta take yang menunggu, dan kapasitasnya sebaiknya dibatasi.
  • Koleksi konkuren menutup persoalan yang sebelumnya ditangani synchronized manual, dengan jaminan yang sudah diuji.

Referensi#