Java: Concurrent Collections and Fork/Join

Description

Sooner or later, several threads need the same collection: a cache that every request reads and some requests fill, a queue of jobs between a web thread and a worker, counts that many threads update, a list of listeners notified on every event. Java's everyday collections (HashMap, ArrayList) aren't built for that; used from several threads at once, they lose data or even throw. The java.util.concurrent package has collections designed for exactly this, plus parallel streams for splitting work across cores.

These tools are safe for what they promise: each single call is safe. The bugs come from assuming more than that, such as two calls in a row being safe together. This skill is about knowing exactly what each one promises.

Here's the path, from easiest to most technical:

1. Why ordinary collections break when shared.
2. Picking the right concurrent collection.
3. ConcurrentHashMap: safe single calls, and the methods that do a whole update.
4. CopyOnWriteArrayList for lists that are read far more than written.
5. BlockingQueue: handing work from one thread to another.
6. Parallel streams: when .parallel() helps and when it hurts.

Why ordinary collections break

A report job tried to speed itself up with a parallel stream:

List<Row> rows = new ArrayList<>();
ids.parallelStream().forEach(id -> rows.add(load(id)));

Some days the report was a few rows short; some days it crashed with ArrayIndexOutOfBoundsException. Several threads were calling ArrayList.add at the same time, and add (check the size, maybe grow the array, store, bump the size) isn't safe if two threads do it at once. With 100,000 numbers, this pattern went wrong in five runs out of five, while collect(Collectors.toList()) returned all 100,000 every time. The same goes for HashMap: shared and written by several threads, it can lose entries.

Picking a concurrent collection

Collections.synchronizedMap(map) also exists: it wraps every call in one lock. It's safe for single calls but slower under load, and you have to hold its lock yourself while iterating.

ConcurrentHashMap

A ConcurrentHashMap lets threads work on different keys at the same time, and reads never wait at all. You can even loop over it while other threads change it: its iterators never throw ConcurrentModificationException (they may or may not show the changes made during the loop).

The rule to remember is single calls are safe; sequences of calls aren't. This count update looks fine and loses updates:

Integer old = counts.get(word);                  // two threads both read 5...
counts.put(word, old == null ? 1 : old + 1);     // ...and both write 6

Between the get and the put, another thread can change the value. ConcurrentHashMap has methods that do the whole update as one step for a key:

counts.merge(word, 1, Integer::sum);                       // add 1, or start at 1
cache.computeIfAbsent(key, k -> load(k));                  // load only if missing
sessions.putIfAbsent(id, newSession);                      // insert only if missing
balances.compute(account, (k, v) -> v == null ? amount : v + amount);

Three more things to know:

• No null keys or values. put(key, null) throws a NullPointerException. In a concurrent map, get returning null has to mean "not there"; a stored null would be ambiguous.
• Keep the functions you pass short, and don't update the same map from inside them. The function runs while part of the map is locked. A memoized Fibonacci that calls computeIfAbsent recursively on the same map fails with IllegalStateException: Recursive update.
• size() is approximate while other threads are writing.

CopyOnWriteArrayList

Every change to a CopyOnWriteArrayList copies the whole underlying array, and every loop over it walks the array as it was when the loop started. So looping never throws and never needs a lock, but each change costs time proportional to the list's size.

That trade is perfect for listener lists: listeners are added rarely, and the list is looped over on every event. It's a poor choice for anything changed often.

BlockingQueue: handing work between threads

A BlockingQueue connects producers (threads that create work) with consumers (threads that do it). Each operation comes in four flavors, depending on what should happen when the queue is full or empty:

Producers usually call put and consumers take, so each side waits when it should. A few rules:

• Bound the queue. ArrayBlockingQueue(1000) holds at most 1,000 items, so a fast producer waits instead of filling memory. LinkedBlockingQueue with no capacity is unbounded: the same trap as a fixed thread pool.
• Give consumers a way to stop. Either interrupt them (take() throws InterruptedException), or send a poison pill: a special item meaning "no more work". Send one pill per consumer, since each consumer stops after taking one.

Parallel streams

Adding .parallel() to a stream splits the work across several threads. It runs on the common pool, the same small, shared pool CompletableFuture uses by default. It helps when all of these are true:

• the data is large (thousands of elements or more),
• the work per element is CPU-heavy (computation, not waiting on I/O),
• the lambdas are independent: they don't change shared state.

It hurts when the work blocks (database calls, HTTP, sleep: that starves the common pool for the whole program), when the data is small, or when the source splits badly (a LinkedList, or Stream.iterate).

And never collect results by changing a shared collection from inside the stream. Use collect, reduce or toList(), which give each thread its own container and merge them at the end:

List<Row> rows = ids.parallelStream().map(this::load).toList();   // safe

Common mistakes

Check yourself

Try to answer before opening each one. Recalling beats re-reading.

1. Why does parallelStream().forEach(list::add) lose elements, while toList() doesn't?

2. Two threads run counts.put(w, counts.get(w) + 1) on a ConcurrentHashMap. Can updates be lost? What's the fix?

3. Why does ConcurrentHashMap refuse null values when HashMap allows them?

4. Three consumer threads read from a BlockingQueue. How many poison pills does the producer send when it's done?

Key takeaways

• Ordinary collections aren't safe to write from several threads; concurrent collections are, one call at a time.
• ConcurrentHashMap: use merge, compute and computeIfAbsent for updates, store no nulls, and keep mapping functions short.
• CopyOnWriteArrayList for rarely changed, often looped lists such as listeners.
• BlockingQueue for handoffs: bounded, with a way to stop consumers (one pill each).
• Parallel streams for large, CPU-heavy, independent work, collected with collect or toList().