Concurrent Collections in Java

Why the ordinary ones are not enough

Map<String, Integer> unsafe = new HashMap<>();
// Concurrent writes can corrupt the internal table during a resize

Map<String, Integer> wrapped = Collections.synchronizedMap(new HashMap<>());
// Every method is atomic, but one lock serialises all access

Map<String, Integer> safe = new ConcurrentHashMap<>();
// Fine grained locking, high throughput, and atomic compound methods

The main choices

ConcurrentHashMap

ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>();

counts.put("java", 1);
counts.merge("java", 1, Integer::sum);              // atomic
counts.computeIfAbsent("sql", key -> load(key));    // atomic, computed once
counts.putIfAbsent("css", 0);                        // atomic
counts.compute("java", (key, value) -> value * 2);   // atomic

counts.forEach(4, (key, value) -> report(key, value));   // parallel above a threshold
int total = counts.reduceValues(4, Integer::sum);

The compound methods are the point. merge, computeIfAbsent and putIfAbsent are single atomic operations, which is what a get followed by a put can never be, no matter how thread safe each call is on its own.

// Still a race, even on a concurrent map
if (!counts.containsKey(key)) {
    counts.put(key, 0);           // another thread may have inserted in between
}

// Correct
counts.putIfAbsent(key, 0);

Rules

• null keys and values are rejected, so get returning null unambiguously means absent.
• Iterators are weakly consistent: they never throw ConcurrentModificationException, and may or may not reflect changes made after they were created.
• size() is an estimate under concurrent modification.
• The function passed to computeIfAbsent must not modify the same map, or it may deadlock.

CopyOnWriteArrayList

List<Listener> listeners = new CopyOnWriteArrayList<>();

listeners.add(listener);        // copies the whole array

for (Listener listener : listeners) {   // iterates a stable snapshot
    listener.onEvent(event);            // safe even if another thread adds one
}

This suits listener lists and configuration snapshots: written rarely, read constantly. It is entirely wrong for a collection that changes often.

Blocking queues

BlockingQueue<Task> queue = new LinkedBlockingQueue<>(1000);

// Producer
queue.put(task);                                   // blocks when full
boolean added = queue.offer(task, 1, TimeUnit.SECONDS);

// Consumer
Task task = queue.take();                          // blocks when empty
Task maybe = queue.poll(500, TimeUnit.MILLISECONDS);

public class Pipeline {

    private final BlockingQueue<Task> queue = new ArrayBlockingQueue<>(100);
    private static final Task POISON = new Task("stop");

    public void produce(List<Task> tasks) throws InterruptedException {
        for (Task task : tasks) {
            queue.put(task);
        }
        queue.put(POISON);          // signal the end
    }

    public void consume() throws InterruptedException {
        Task task;
        while ((task = queue.take()) != POISON) {
            process(task);
        }
    }
}

A bounded blocking queue also provides back pressure: when consumers fall behind, producers block instead of filling memory.

Coordination utilities

CountDownLatch ready = new CountDownLatch(3);      // wait for three tasks
ready.countDown();
ready.await();                                      // one use only

CyclicBarrier barrier = new CyclicBarrier(4, this::onAllArrived);   // reusable
barrier.await();

Semaphore permits = new Semaphore(5);               // limit concurrent access
permits.acquire();
try {
    callRateLimitedApi();
} finally {
    permits.release();
}

Choosing

Common mistakes

• Assuming a concurrent collection makes a check then act sequence atomic.
• Putting a null into a ConcurrentHashMap.
• Using CopyOnWriteArrayList for a frequently modified list.
• Reusing a CountDownLatch, which cannot be reset.
• Using an unbounded queue and losing back pressure.
• Modifying a ConcurrentHashMap from inside its own computeIfAbsent function.

Best practices

• Prefer a concurrent collection to a synchronised wrapper.
• Use the atomic compound methods rather than composing several calls.
• Bound your queues.
• Choose the implementation from the read to write ratio.
• Release a semaphore permit in a finally block.
• Remember that size() is approximate under concurrency.

Practice

1. Count words from several threads with ConcurrentHashMap.merge.
2. Explain why containsKey followed by put is still a race.
3. Build a producer and consumer pipeline with a bounded queue and a stop signal.
4. When is CopyOnWriteArrayList the right choice, and when is it clearly wrong?
5. Limit concurrent calls to an external service to five using a Semaphore.

Conclusion

The concurrent collections are designed for shared access rather than retrofitted with a lock. Use their atomic compound methods, bound your queues for back pressure, and choose the implementation from how the data is actually read and written.