Concurrent Collections

The collections in java.util.concurrent are safe to use from many threads without external locking, and each one makes a different trade between read cost, write cost, and blocking. This lesson explains how ConcurrentHashMap works since Java 8, when to use the other concurrent collections, and ends with a multi-stage producer-consumer pipeline. It uses executors from Executors and CompletableFuture.

Collection	Model	Typical use
`ConcurrentHashMap`	CAS plus per-bin locks	Shared caches, counters by key
`CopyOnWriteArrayList`	Copy the array on every write	Listener lists, rarely changed config
`ArrayBlockingQueue`	One lock, bounded array	Bounded hand-off between stages
`LinkedBlockingQueue`	Two locks (put and take), optionally bounded	Higher-throughput hand-off
`SynchronousQueue`	No capacity, direct hand-off	Cached thread pools
`DelayQueue`	Priority by delay	Timers, retries, expiry
`ConcurrentLinkedQueue`	Lock-free linked list (Michael-Scott)	Non-blocking queue
`ConcurrentSkipListMap`	Lock-free skip list	Sorted concurrent map, range queries

Example 1: Why a synchronized wrapper is not enough

Collections.synchronizedMap locks every single call, but a check-then-act sequence of two calls is still a race: another thread can run between get and put. Concurrent maps offer atomic compound operations instead.

Java

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class WordCount {
    public static void main(String[] args) {
        Map<String, Integer> counts = new ConcurrentHashMap<>();
        String[] words = {"buy", "sell", "buy", "hold", "buy", "sell"};

        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            for (int rep = 0; rep < 1000; rep++) {
                for (String w : words) {
                    pool.execute(() -> counts.merge(w, 1, Integer::sum));   // atomic per key
                }
            }
        }                                                                   // close() waits
        System.out.println(new java.util.TreeMap<>(counts));
    }
}

Output

{buy=3000, hold=1000, sell=2000}

The broken version counts.put(w, counts.getOrDefault(w, 0) + 1) loses updates when two threads read the same old value. merge reads, combines, and writes as one atomic step for that key. The result is copied into a TreeMap only to print keys in a fixed order.

Example 2: How ConcurrentHashMap is built (Java 8 and later)

Java 7 split the map into 16 Segments, each a lock. Java 8 replaced that with a single table of bins:

Empty bin insert: the first node is placed with a compare-and-set (CAS) on the table slot, with no lock.
Non-empty bin: the writer locks the first node of that bin with synchronized and appends or updates. Writers to different bins do not contend.
Reads take no lock. Node val and next fields are volatile, and table slots are read with volatile semantics, so a reader sees a complete node.
Treeified bins: when a bin reaches 8 nodes (TREEIFY_THRESHOLD) and the table has at least 64 slots (MIN_TREEIFY_CAPACITY), the bin becomes a red-black tree (TreeBin), so a bin with many colliding keys costs O(log n) instead of O(n). It turns back into a list at 6 nodes (UNTREEIFY_THRESHOLD). If the table is smaller than 64, it resizes instead.
Resize: a ForwardingNode marks moved bins, and threads that call put during a resize help transfer bins.
Nulls: neither keys nor values may be null, because get returning null must mean "absent" without a second, racy containsKey call.

Java

import java.util.concurrent.ConcurrentHashMap;

public class ChmBasics {
    record BadKey(int id) {
        @Override public int hashCode() { return 42; }      // every key collides
    }

    public static void main(String[] args) {
        ConcurrentHashMap<BadKey, Integer> map = new ConcurrentHashMap<>(128);
        for (int i = 0; i < 20; i++) map.put(new BadKey(i), i);
        System.out.println("size = " + map.size() + ", get(7) = " + map.get(new BadKey(7)));

        try {
            map.put(new BadKey(99), null);
        } catch (NullPointerException e) {
            System.out.println("null value rejected");
        }

        System.out.println("putIfAbsent existing: " + map.putIfAbsent(new BadKey(1), 100));
        System.out.println("putIfAbsent new: " + map.putIfAbsent(new BadKey(50), 50));
    }
}

Output

size = 20, get(7) = 7
null value rejected
putIfAbsent existing: 1
putIfAbsent new: null

All 20 keys land in one bin, which becomes a tree once it passes the threshold. Because BadKey is a record (not Comparable), the tree orders ties by class name and identity hash, so lookups still work but are slower than with comparable keys. putIfAbsent returns the previous value, or null when it inserted.

Example 3: compute, computeIfAbsent, and size counting

compute, computeIfAbsent, computeIfPresent, and merge run their function while holding the bin lock, so the update is atomic for that key. Two consequences follow: the function must be short, and it must not modify the same map, because that can deadlock or throw IllegalStateException("Recursive update").

Java

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

public class ComputeDemo {
    public static void main(String[] args) {
        ConcurrentHashMap<String, List<Integer>> ordersByUser = new ConcurrentHashMap<>();
        ConcurrentHashMap<String, LongAdder> hits = new ConcurrentHashMap<>();
        AtomicInteger factoryCalls = new AtomicInteger();

        try (ExecutorService pool = Executors.newFixedThreadPool(8)) {
            for (int i = 0; i < 1000; i++) {
                int order = i;
                String user = "u" + (i % 3);
                pool.execute(() -> {
                    ordersByUser.computeIfAbsent(user, k -> {
                        factoryCalls.incrementAndGet();
                        return new CopyOnWriteArrayList<>();
                    }).add(order);
                    hits.computeIfAbsent(user, k -> new LongAdder()).increment();
                });
            }
        }
        System.out.println("factory calls = " + factoryCalls.get());
        System.out.println("u0 orders = " + ordersByUser.get("u0").size());
        System.out.println("u0 hits = " + hits.get("u0").sum());
        System.out.println("mappingCount = " + ordersByUser.mappingCount());
    }
}

Output

factory calls = 3
u0 orders = 334
u0 hits = 334
mappingCount = 3

The factory runs exactly once per key even with 8 threads racing, because computeIfAbsent is atomic. The value list must itself be thread-safe, because add happens outside the map's lock. LongAdder is a counter that spreads increments across cells; it is covered in Atomic Classes and CAS.

The map counts its own size the same way. It keeps a baseCount updated by CAS, and when that CAS fails under contention, threads add to an array of CounterCells instead. size() and mappingCount() sum baseCount and all cells. The sum is not taken under a lock, so during concurrent updates it is an estimate; it is exact once writers stop. mappingCount() returns a long and is preferred because size() is capped at Integer.MAX_VALUE.

Example 4: CopyOnWriteArrayList and weakly consistent iteration

CopyOnWriteArrayList copies its whole backing array on every write. Reads and iteration use the array snapshot they started with and never lock. This fits lists that are read very often and changed rarely, such as event listeners.

Java

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

public class Snapshots {
    public static void main(String[] args) {
        List<String> listeners = new CopyOnWriteArrayList<>(List.of("audit", "metrics"));
        for (String l : listeners) {
            if (l.equals("audit")) listeners.add("risk");    // no ConcurrentModificationException
            System.out.println("notify " + l);
        }
        System.out.println("now " + listeners);

        Iterator<String> it = listeners.iterator();
        try {
            it.next();
            it.remove();
        } catch (UnsupportedOperationException e) {
            System.out.println("iterator remove unsupported");
        }

        ConcurrentHashMap<Integer, String> m = new ConcurrentHashMap<>();
        for (int i = 0; i < 3; i++) m.put(i, "v" + i);
        for (Integer k : m.keySet()) m.remove(k);            // allowed, weakly consistent
        System.out.println("map empty: " + m.isEmpty());
    }
}

Output

notify audit
notify metrics
now [audit, metrics, risk]
iterator remove unsupported
map empty: true

The loop does not see risk because the iterator holds the old array. Each write costs O(n) time and allocates a new array, so a list with frequent writes or many elements is a poor fit. ConcurrentHashMap iterators are weakly consistent: they never throw ConcurrentModificationException, they reflect some state at or after creation, and they may or may not show changes made during iteration.

Example 5: The BlockingQueue family

A BlockingQueue adds methods that wait. The four method groups differ in what they do when the queue is full or empty:

Action	Throws	Returns special value	Blocks	Times out
Insert	`add`	`offer` (false)	`put`	`offer(e, t, unit)`
Remove	`remove`	`poll` (null)	`take`	`poll(t, unit)`

Java

import java.util.concurrent.*;

public class Queues {
    record Retry(String id, long dueNanos) implements Delayed {
        public long getDelay(TimeUnit unit) {
            return unit.convert(dueNanos - System.nanoTime(), TimeUnit.NANOSECONDS);
        }
        public int compareTo(Delayed o) {
            return Long.compare(dueNanos, ((Retry) o).dueNanos);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        System.out.println(abq.offer(1) + " " + abq.offer(2) + " " + abq.offer(3));
        System.out.println("poll with timeout: " + new LinkedBlockingQueue<Integer>()
                .poll(10, TimeUnit.MILLISECONDS));

        SynchronousQueue<String> sq = new SynchronousQueue<>();
        System.out.println("offer with no taker: " + sq.offer("x"));
        Thread taker = new Thread(() -> {
            try { System.out.println("took " + sq.take()); } catch (InterruptedException ignored) {}
        });
        taker.start();
        sq.put("order-1");                                   // waits for the taker
        taker.join();

        DelayQueue<Retry> dq = new DelayQueue<>();
        long now = System.nanoTime();
        dq.put(new Retry("late", now + TimeUnit.MILLISECONDS.toNanos(40)));
        dq.put(new Retry("early", now + TimeUnit.MILLISECONDS.toNanos(20)));
        System.out.println("ready now: " + dq.poll());
        System.out.println(dq.take().id() + ", " + dq.take().id());
    }
}

Output

true true false
poll with timeout: null
offer with no taker: false
took order-1
ready now: null
early, late

Differences that matter:

ArrayBlockingQueue is always bounded and uses one ReentrantLock for puts and takes. It allocates nothing per element.
LinkedBlockingQueue uses separate put and take locks, so a producer and a consumer do not block each other. Its default capacity is Integer.MAX_VALUE, which is unbounded in practice. It allocates a node per element.
SynchronousQueue holds nothing; each put waits for a take.
DelayQueue releases an element only after its delay expires, in order of expiry.

Example 6: ConcurrentLinkedQueue and ConcurrentSkipListMap

ConcurrentLinkedQueue is an unbounded, non-blocking FIFO queue based on the Michael-Scott lock-free algorithm: it updates head and tail with CAS. poll returns null when empty instead of waiting. Its size() walks the whole list, so it is O(n).

ConcurrentSkipListMap is a sorted concurrent map built on a skip list (linked lists in levels, where higher levels skip over more nodes), giving expected O(log n) operations without locks. It implements ConcurrentNavigableMap, so range queries work concurrently. A concurrent tree map does not exist in the JDK because rebalancing a tree touches many nodes at once.

Java

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SkipListBook {
    public static void main(String[] args) {
        ConcurrentLinkedQueue<String> events = new ConcurrentLinkedQueue<>();
        ConcurrentSkipListMap<Long, Integer> bids = new ConcurrentSkipListMap<>(java.util.Comparator.reverseOrder());

        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            for (long price = 100; price < 110; price++) {
                long p = price;
                pool.execute(() -> {
                    bids.merge(p, 10, Integer::sum);
                    events.add("bid@" + p);
                });
            }
        }
        System.out.println("events = " + events.size());
        System.out.println("best bid = " + bids.firstKey());
        System.out.println("top 3 = " + bids.headMap(107L, true));
        System.out.println("bids >= 108 = " + bids.headMap(107L).keySet());
        System.out.println("poll empty: " + new ConcurrentLinkedQueue<String>().poll());
    }
}

Output

events = 10
best bid = 109
top 3 = {109=10, 108=10, 107=10}
bids >= 108 = [109, 108]
poll empty: null

With a reversed comparator, the "head" of the map is the highest price, so headMap(107L, true) returns the three best bids, 109 down to 107 inclusive. headMap(107L) excludes 107, leaving 109 and 108. This descending sorted view is how an order book side is often modeled.

Example 7: Project: multi-stage producer-consumer pipeline

The pipeline has three stages connected by bounded queues: a parser reads raw order lines, two validators check them, and a writer aggregates totals. Bounded queues give back-pressure: a fast stage blocks on put when the next stage falls behind. Shutdown uses a poison pill, a special element that tells a consumer to stop, sent once per consumer.

Java

import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;

public class Pipeline {
    record Order(int id, String symbol, int qty) {}
    static final Order POISON = new Order(-1, "", 0);

    public static void main(String[] args) throws Exception {
        BlockingQueue<String> raw = new ArrayBlockingQueue<>(16);
        BlockingQueue<Order> valid = new ArrayBlockingQueue<>(16);
        ConcurrentHashMap<String, LongAdder> qtyBySymbol = new ConcurrentHashMap<>();
        LongAdder rejected = new LongAdder();
        int validators = 2;

        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            // Stage 1: producer
            pool.execute(() -> {
                try {
                    for (int i = 1; i <= 1000; i++) {
                        String sym = (i % 2 == 0) ? "AAPL" : "MSFT";
                        int qty = (i % 10 == 0) ? -5 : 10;        // every 10th line is invalid
                        raw.put(i + "," + sym + "," + qty);
                    }
                    for (int v = 0; v < validators; v++) raw.put("EOF");
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });

            // Stage 2: validators
            CountDownLatch validatorsDone = new CountDownLatch(validators);
            for (int v = 0; v < validators; v++) {
                pool.execute(() -> {
                    try {
                        for (String line; !(line = raw.take()).equals("EOF"); ) {
                            String[] f = line.split(",");
                            Order o = new Order(Integer.parseInt(f[0]), f[1], Integer.parseInt(f[2]));
                            if (o.qty() > 0) valid.put(o); else rejected.increment();
                        }
                    } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                    finally { validatorsDone.countDown(); }
                });
            }

            // Stage 3: writer
            pool.execute(() -> {
                try {
                    for (Order o; (o = valid.take()) != POISON; ) {
                        qtyBySymbol.computeIfAbsent(o.symbol(), k -> new LongAdder()).add(o.qty());
                    }
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });

            validatorsDone.await();
            valid.put(POISON);                          // only after all validators finished
        }

        System.out.println("AAPL qty = " + qtyBySymbol.get("AAPL").sum());
        System.out.println("MSFT qty = " + qtyBySymbol.get("MSFT").sum());
        System.out.println("rejected = " + rejected.sum());
    }
}

Output

AAPL qty = 4000
MSFT qty = 5000
rejected = 100

Of 1000 lines, the 100 lines with i % 10 == 0 are rejected, and all of them have even i, so they are AAPL lines. AAPL keeps 500 − 100 = 400 orders of 10 (4000), MSFT keeps 500 (5000). The writer gets its pill only after the latch shows both validators are done; sending it earlier could stop the writer while orders are still arriving. The pool has exactly 4 threads for the 4 long-running tasks; with fewer, a stage would never start and the pipeline would deadlock.

Interview points

ConcurrentHashMap (Java 8+): CAS into empty bins, synchronized on the bin head otherwise, lock-free reads, cooperative resizing.
Bins treeify at 8 nodes when the table has at least 64 slots, and untreeify at 6.
compute* and merge are atomic per key and run under the bin lock; keep functions short and do not touch the same map inside them.
size() sums baseCount and CounterCells and is an estimate during updates; prefer mappingCount().
No null keys or values, so get returning null means absent.
CopyOnWriteArrayList: snapshot iteration, O(n) writes; fits read-mostly lists.
ArrayBlockingQueue has one lock; LinkedBlockingQueue has two and is unbounded by default.
ConcurrentSkipListMap is the sorted concurrent map; ConcurrentLinkedQueue.size() is O(n).
Pipelines use bounded queues for back-pressure and one poison pill per consumer.

Summary

Concurrent collections replace check-then-act races with atomic compound methods.
ConcurrentHashMap locks per bin, not per map.
Iterators of concurrent collections are weakly consistent or snapshot-based.
BlockingQueue methods throw, return a special value, block, or time out.
SynchronousQueue has no capacity; DelayQueue releases elements after their delay.
A pipeline stops cleanly when each stage forwards a stop signal after its inputs are drained.