Concurrent Collections and Utilities

Java provides a rich set of thread-safe collections and synchronization utilities in the java.util.concurrent package to simplify concurrent programming and improve performance.

---

Introduction

Traditional synchronized collections (e.g., Collections.synchronizedList) provide basic thread safety but can be inefficient under high contention. The java.util.concurrent package offers scalable, lock-free, and advanced thread-safe collections and synchronization utilities.

---

Thread-Safe Collections

ConcurrentHashMap

• Highly concurrent, lock-splitting map
• Allows concurrent reads and updates
• No null keys or values

Map<String, Integer> map = new ConcurrentHashMap<>();
map.put("A", 1);
int value = map.get("A");
map.computeIfAbsent("B", k -> 42);

CopyOnWriteArrayList / CopyOnWriteArraySet

• Good for lists/sets with frequent reads and infrequent writes
• Writes create a new copy of the underlying array

List<String> list = new CopyOnWriteArrayList<>();
list.add("A");
for (String s : list) { /* safe iteration */ }

ConcurrentSkipListMap / ConcurrentSkipListSet

• Scalable, sorted, thread-safe maps/sets
• Based on skip list data structure

ConcurrentSkipListMap<String, Integer> sortedMap = new ConcurrentSkipListMap<>();

Synchronized Wrappers

• Legacy approach: Collections.synchronizedList, etc.
• Less scalable than concurrent collections

List<String> syncList = Collections.synchronizedList(new ArrayList<>());

---

Blocking Queues

Blocking queues are thread-safe queues that block on insertion/removal when full/empty.

• ArrayBlockingQueue: Fixed-size, array-backed
• LinkedBlockingQueue: Optionally bounded, linked nodes
• PriorityBlockingQueue: Priority ordering, unbounded
• DelayQueue: Elements become available after a delay
• SynchronousQueue: No internal capacity, handoff
• LinkedTransferQueue: Advanced transfer capabilities

BlockingQueue<String> queue = new LinkedBlockingQueue<>();
queue.put("item"); // blocks if full
String item = queue.take(); // blocks if empty

---

Atomic Variables and Classes

Lock-free, thread-safe variables for single-value updates.

• AtomicInteger, AtomicLong: Atomic numeric operations
• AtomicReference: Atomic object reference
• AtomicStampedReference: Reference + version (avoids ABA problem)
• AtomicIntegerArray, AtomicLongArray, AtomicReferenceArray

AtomicInteger count = new AtomicInteger(0);
count.incrementAndGet();
int value = count.get();

---

Synchronization Utilities

CountDownLatch

• Allows threads to wait until a set of operations completes

CountDownLatch latch = new CountDownLatch(3);
// Each worker calls latch.countDown();
latch.await(); // Main thread waits

CyclicBarrier

• Allows threads to wait for each other to reach a common barrier point

CyclicBarrier barrier = new CyclicBarrier(3);
barrier.await(); // Each thread waits until all arrive

Semaphore

• Controls access to a resource with a set number of permits

Semaphore semaphore = new Semaphore(2);
semaphore.acquire();
// Access resource
semaphore.release();

Phaser

• Flexible alternative to CountDownLatch and CyclicBarrier for dynamic parties

Phaser phaser = new Phaser(3);
phaser.arriveAndAwaitAdvance();

Exchanger

• Allows two threads to exchange objects

Exchanger<String> exchanger = new Exchanger<>();
String received = exchanger.exchange("data");

---

Best Practices

• Prefer concurrent collections over synchronized wrappers
• Use blocking queues for producer-consumer scenarios
• Use atomic variables for lock-free single-value updates
• Use synchronization utilities to coordinate thread progress
• Choose the right tool for the workload (e.g., CopyOnWrite for mostly-read lists)
• Avoid mixing manual synchronization with concurrent collections

---

Code Examples

Producer-Consumer with BlockingQueue

BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

// Producer
new Thread(() -> {
    try {
        for (int i = 0; i < 10; i++) {
            queue.put(i);
            System.out.println("Produced: " + i);
            Thread.sleep(100);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

// Consumer
new Thread(() -> {
    try {
        while (true) {
            Integer item = queue.take();
            System.out.println("Consumed: " + item);
            Thread.sleep(200);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

AtomicInteger Usage

AtomicInteger counter = new AtomicInteger(0);
List<Thread> threads = new ArrayList<>();

for (int i = 0; i < 100; i++) {
    Thread t = new Thread(() -> {
        for (int j = 0; j < 1000; j++) {
            counter.incrementAndGet();
        }
    });
    threads.add(t);
    t.start();
}

// Wait for all threads to complete
for (Thread t : threads) {
    t.join();
}
System.out.println("Final count: " + counter.get()); // Should be 100,000

Using CountDownLatch

CountDownLatch latch = new CountDownLatch(3);
List<Thread> workers = new ArrayList<>();

for (int i = 0; i < 3; i++) {
    final int workerId = i;
    Thread worker = new Thread(() -> {
        try {
            // Simulate work
            System.out.println("Worker " + workerId + " starting");
            Thread.sleep(1000 + new Random().nextInt(2000));
            System.out.println("Worker " + workerId + " finished");
            latch.countDown();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
    workers.add(worker);
    worker.start();
}

// Main thread waits for all workers
System.out.println("Waiting for workers to finish...");
latch.await();
System.out.println("All workers finished!");