Concurrent Collections in Java

The concurrent collections in java.util.concurrent are collections designed to be used by many threads at once, without locking the whole collection. They are faster than the synchronized wrappers from "Synchronized vs Concurrent Collections", offer atomic compound operations such as merge and putIfAbsent, and have iterators that never throw ConcurrentModificationException. In this lesson, you will learn the main concurrent collections, how ConcurrentHashMap works inside, and the blocking queues used for passing work between threads.

The Main Concurrent Collections

How ConcurrentHashMap Works Inside

• Reads (get) never lock; they read volatile node fields.
• A write to an empty bucket places the new node with one CAS.
• A write to a busy bucket locks only that bucket, so writes to other buckets run in parallel.
• Resizing is shared: threads that arrive during a resize help move buckets.
• Like HashMap, long buckets become trees. Unlike HashMap, it rejects null keys and values.

Example: Counting Words from Many Threads

Let's see an example where four threads count words into one ConcurrentHashMap.

import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class WordCounts {
    public static void main(String[] args) {
        List<String> pages = List.of(
            "java threads java locks",
            "java streams maps",
            "threads maps maps",
            "locks java");
        ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>();
 
        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            for (String page : pages) {
                pool.submit(() -> {
                    for (String w : page.split(" ")) {
                        counts.merge(w, 1, Integer::sum);
                    }
                });
            }
        }
        System.out.println(new TreeMap<>(counts));
 
        counts.computeIfAbsent("records", k -> 0);
        int total = counts.reduceValues(1, Integer::sum);
        System.out.println("Words counted: " + total);
        System.out.println("Most frequent: " + counts.reduceEntries(1,
            (a, b) -> a.getValue() >= b.getValue() ? a : b).getKey());
    }
}

{java=4, locks=2, maps=3, streams=1, threads=2}
Words counted: 12
Most frequent: java

Explanation: merge(w, 1, Integer::sum) is atomic: even when two threads count the same word at the same time, neither update is lost. After printing, computeIfAbsent adds the key "records" with 0, only because it was missing, also atomically; it doesn't change the total. reduceValues and reduceEntries are bulk operations that can run in parallel (the first argument is a parallelism threshold).

Example: BlockingQueue and CopyOnWriteArrayList

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
 
public class QueuesAndLists {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> jobs = new LinkedBlockingQueue<>(2);
        System.out.println("offer 1: " + jobs.offer("print report"));
        System.out.println("offer 2: " + jobs.offer("send email"));
        System.out.println("offer 3 (full): " + jobs.offer("backup"));
        System.out.println("take: " + jobs.take());
        System.out.println("poll with timeout: "
            + jobs.poll(100, TimeUnit.MILLISECONDS));
        System.out.println("poll on empty: "
            + jobs.poll(100, TimeUnit.MILLISECONDS));
 
        List<String> listeners = new CopyOnWriteArrayList<>(
            List.of("Navika", "Srihitha"));
        for (String l : listeners) {
            listeners.add("Advitha");
        }
        System.out.println("Listeners: " + listeners);
    }
}

offer 1: true
offer 2: true
offer 3 (full): false
take: print report
poll with timeout: send email
poll on empty: null
Listeners: [Navika, Srihitha, Advitha, Advitha]

Explanation: A BlockingQueue with capacity 2 accepts two jobs; offer returns false for the third instead of waiting (put would wait). take and poll remove jobs in order, and poll with a timeout returns null when nothing arrives. Adding to a CopyOnWriteArrayList while looping over it doesn't throw: the loop reads its snapshot of two elements, so two "Advitha" entries are added.

BlockingQueue Methods

Common Mistakes

if (!map.containsKey(k)) map.put(k, v) on a concurrent map. Use putIfAbsent or computeIfAbsent.

Storing null in ConcurrentHashMap. It throws NullPointerException.

Using CopyOnWriteArrayList for lists that change often. Every write copies the whole array.

Interview Corner

Practice

1. Easy: Count how many times each letter appears in four strings, using four threads and ConcurrentHashMap.merge.
2. Harder: Build a simple job system: two producer threads put jobs into a LinkedBlockingQueue, three consumer threads take and process them, and a poison pill stops the consumers.

Key Takeaways