並行コレクションの概要

並行コレクションとは、マルチスレッド環境でのデータアクセスを安全かつ効率的に処理するために設計されたJava標準ライブラリの一部です。通常のコレクションクラス（ArrayListやHashMapなど）は単一スレッド向けに最適化されていますが、これらを複数スレッドから同時に操作すると、データの一貫性が失われる可能性があります。並行コレクションはこのような問題に対応するため、同期機構を内蔵しており、パフォーマンスと安全性を両立させています。

主要な並行コレクションとその特徴

ConcurrentHashMap

ConcurrentHashMapは、HashMapのスレッドセーフ版であり、高負荷な同時アクセスにも対応しています。JDK 1.7まではセグメントロック方式を採用していましたが、JDK 1.8以降ではCAS（Compare-And-Swap）操作と連結リストまたは赤黒木を使用することで、より高い並列性を実現しています。これにより、異なるキーに対する同時更新が可能となり、全体的なスループットが向上します。

CopyOnWriteArrayList

CopyOnWriteArrayListは、読み取り頻度が高いが書き込み頻度が低いシナリオに適したリスト構造です。要素の追加・削除時には元の配列をコピーし、新しい配列上で変更を行った後、参照を差し替えます。この仕組みにより、読み取り処理は常に古い配列に対して行われるため、ロックフリーで高速なアクセスが可能です。

ConcurrentLinkedQueue

ConcurrentLinkedQueueは、無制限のFIFOキューとして機能し、ノードベースの連結リストで実装されています。このクラスはCAS操作によって非ブロッキングなスレッドセーフを実現しており、大量のスレッドによるエンキューおよびデキュー操作を効率的に処理できます。

実際の利用例

ConcurrentHashMapの利用

import java.util.concurrent.ConcurrentHashMap;

public class CHMUsage {
    private static final ConcurrentHashMap<String, Long> counter = new ConcurrentHashMap<>();

    public static void increment(String key) {
        counter.compute(key, (k, v) -> (v == null) ? 1L : v + 1);
    }

    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 500; i++) increment("pageA");
        });
        
        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 500; i++) increment("pageA");
        });

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

        System.out.println("Total visits: " + counter.get("pageA"));
    }
}

CopyOnWriteArrayListの利用

import java.util.Iterator;
import java.util.concurrent.CopyOnWriteArrayList;

public class COWALUsage {
    private static final CopyOnWriteArrayList<String> observers = new CopyOnWriteArrayList<>();

    public static void addObserver(String observer) {
        observers.add(observer);
    }

    public static void notifyAllObservers(String message) {
        for (String obs : observers) {
            System.out.println(obs + " received: " + message);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        addObserver("Observer1");
        addObserver("Observer2");

        Thread reader = new Thread(() -> {
            Iterator<String> iter = observers.iterator();
            while(iter.hasNext()) {
                System.out.println("Current observer: " + iter.next());
                try { Thread.sleep(100); } catch(InterruptedException ignored){}
            }
        });

        Thread writer = new Thread(() -> observers.add("Observer3"));

        reader.start();
        writer.start();
        reader.join();
        writer.join();
    }
}

ConcurrentLinkedQueueの利用

import java.util.concurrent.ConcurrentLinkedQueue;

public class CLQUsage {
    private static final ConcurrentLinkedQueue<String> taskQueue = new ConcurrentLinkedQueue<>();

    public static void submitTask(String task) {
        taskQueue.offer(task);
    }

    public static String fetchTask() {
        return taskQueue.poll();
    }

    public static void main(String[] args) throws InterruptedException {
        Thread producer1 = new Thread(() -> {
            for(int i=0; i<5; i++) submitTask("Task-P1-" + i);
        });

        Thread producer2 = new Thread(() -> {
            for(int i=0; i<5; i++) submitTask("Task-P2-" + i);
        });

        Thread consumer1 = new Thread(() -> {
            String task;
            while((task = fetchTask()) != null)
                System.out.println("Consumer1 processing: " + task);
        });

        Thread consumer2 = new Thread(() -> {
            String task;
            while((task = fetchTask()) != null)
                System.out.println("Consumer2 processing: " + task);
        });

        producer1.start(); producer2.start();
        consumer1.start(); consumer2.start();

        producer1.join(); producer2.join();
        consumer1.join(); consumer2.join();
    }
}

適切なコレクションの選択と性能調整

並行コレクションを選ぶ際には、アプリケーションの特性に基づいた判断が必要です。例えば、頻繁な読取りと稀な書込みがある場合はCopyOnWriteArrayListが有効であり、頻繁な更新と検索が必要であればConcurrentHashMapが適しています。また、初期キャパシティやロードファクターなどのパラメータを適切に設定することで、メモリ使用量やパフォーマンスをさらに改善することが可能です。