Java Concurrent Collections and Parallel Streams (1Z0-831)

Ordinary java.util collections such as ArrayList, LinkedList, HashMap and TreeMap are not thread-safe: if two threads change one at the same time, updates can be lost or the internal structure corrupted, and if the collection changes while it is being iterated, its fail-fast iterator throws ConcurrentModificationException on the next call. Java gives you two families of thread-safe replacements, and they behave very differently.

The key lesson is that synchronized is not the same as concurrent. A Vector or a synchronized wrapper makes each call atomic, but its iterator is the same fail-fast iterator as the plain collection's. Removing an element inside a for-each loop over any of them is detected at the next call to next():

List<Integer> list = new ArrayList<>(List.of(1, 2, 3, 4));
for (Integer i : list) {
    if (i == 1) {
        list.remove(i);   // next iteration: ConcurrentModificationException
    }
}

The same loop over a Vector, a Collections.synchronizedList, a LinkedList or a HashMap's key set can throw the same way; detection is best effort, so a removal that leaves the loop with nothing left to visit (for example removing the second-to-last element) ends the loop quietly instead. Never rely on the exception, and never rely on its absence. Over a CopyOnWriteArrayList, a ConcurrentSkipListSet or a ConcurrentHashMap key set it simply works. A weakly consistent iterator (the concurrent maps, sets and queues) never throws, traverses each element at most once, and may or may not reflect changes made after it was created. A snapshot iterator (the copy-on-write classes) sees exactly the contents at the moment it was created.

Collections.synchronizedList, synchronizedSet, synchronizedMap (and the sorted/navigable variants) return a wrapper whose every method is synchronized on one mutex, which by default is the wrapper object itself. That makes each individual call thread-safe: concurrent add or put calls from many threads are all kept. It does not make a sequence of calls atomic, and it does not protect traversal.

• Iteration needs a manual lock on the wrapper. The Javadoc requires you to synchronize on the wrapper when you traverse it with an Iterator, a for-each loop, a Spliterator or a Stream. For a synchronized map that includes its keySet(), values() and entrySet() views: lock the map, not the view. Locking any other object protects nothing, because the wrapper's methods never take that lock.
• The wrapper's own bulk methods are already locked. forEach, removeIf and replaceAll on the wrapper run the whole traversal inside its mutex. A stream or an explicit iterator obtained from it does not.
• Check-then-act needs one lock around both steps. A test followed by an update is two separately synchronized calls; another thread can act between them.

Map<String, Integer> stock =
    Collections.synchronizedMap(new HashMap<>());

synchronized (stock) {            // lock the wrapper itself
    if (!stock.containsKey("ink")) {
        stock.put("ink", 1);      // check and act are now one step
    }
}

synchronized (stock) {            // required to walk a view
    for (String k : stock.keySet()) {
        System.out.println(k);
    }
}

stock.forEach((k, v) -> System.out.println(k)); // locked inside

Switching to another thread-safe type does not fix check-then-act by itself: the same two-step test on a concurrent collection is still two steps. What fixes it is either one lock held around both steps, as above, or a single method that checks and updates atomically, such as putIfAbsent on a concurrent map or add on a concurrent set (which reports whether the element was new). Visibility tools such as volatile do not make two calls atomic; locks and atomic variables are covered in the objective 8.2 lesson.

ConcurrentHashMap is the default thread-safe Map: reads do not block, writes lock only a small part of the table, and its key, value and entry views have weakly consistent iterators, so you can remove entries while looping over keySet() without an exception. Three behaviours are worth knowing precisely.

1. It rejects null

The rule is general: wherever the map must look up a null key or store a null value, it throws NullPointerException — in storing, reading and searching methods alike. The reason is that get returning null must mean only one thing: no mapping. ConcurrentSkipListMap and the other concurrent maps behave the same way. A few methods take a value only to compare against or to hand back unchanged; they tolerate a null there, finding no match or returning it, rather than throwing. Read each call and ask whether the map would have to hash or store the null.

A null returned by a function is different: it means "no mapping". computeIfAbsent(k, f) adds nothing when f returns null; compute, computeIfPresent or merge whose function returns null removes the entry (or, for an absent key, adds nothing).

Map<String, Integer> m = new ConcurrentHashMap<>();
m.put("p", 5);
m.computeIfPresent("p", (k, v) -> null); // removes "p"
m.compute("q", (k, v) -> null);         // adds nothing
System.out.println(m);                   // {}
m.get(null);                             // NullPointerException

2. Single methods are atomic; get-then-put is not

Each call on a ConcurrentHashMap is atomic, but two calls are two steps: any update that reads the current value in one call and writes the new value in another lets two threads read the same value and both write the same result, losing an update, whichever methods do the reading and the writing. The read-modify-write methods run the whole update atomically for that key:

Map<Integer, Integer> byLen = new ConcurrentHashMap<>();
Stream.of("ab", "abc", "ab", "cd").parallel()
    .forEach(w -> byLen.merge(w.length(), 1, Integer::sum));
System.out.println(byLen.get(2) + " " + byLen.get(3)); // 3 1

Read what each method actually does before trusting it. computeIfAbsent never calls its function when the key already has a value, so it can create an entry but never update one; and the atomicity covers only the call itself, not extra reads of the map made inside the function. A common safe idiom for building groups from many threads is to let computeIfAbsent create a thread-safe container once per key and then add to it:

Map<String, Set<String>> team = new ConcurrentHashMap<>();
team.computeIfAbsent("ops", d -> ConcurrentHashMap.newKeySet())
    .add("Ana");                 // container created once, atomically
System.out.println(team);        // {ops=[Ana]}

The container must itself be thread-safe (a concurrent set, a CopyOnWriteArrayList, a queue), because several threads may add to it at once.

3. A thread-safe Set comes from the map

There is no ConcurrentHashSet class. ConcurrentHashMap.newKeySet() returns a Set (a KeySetView) backed by a ConcurrentHashMap: hash-based, unordered, cheap writes, no need for Comparable, weakly consistent iteration. size() returns an int; mappingCount() returns a long estimate for very large maps.

4. Bulk operations: forEach, search and reduce

ConcurrentHashMap has its own bulk methods that can process the map in parallel without a stream. Each takes a long parallelismThreshold first: if the map has fewer elements than the threshold, the operation runs sequentially; 1 allows maximal parallelism and Long.MAX_VALUE forces sequential execution. The threshold changes only how the work is split, never the result of a correct operation.

• forEach(threshold, (k, v) -> ...) — run an action for every entry (order not specified).
• search(threshold, (k, v) -> ...) — return the first non-null result the function produces on any thread, then stop; null if every call returned null. When several entries match, which one you get is not specified.
• reduce(threshold, transformer, reducer) and the reduceKeys, reduceValues, reduceEntries variants — transform each element, then combine the results. A transformer that returns null skips that element; if every element is skipped the result is null.

ConcurrentHashMap<String, Integer> c = new ConcurrentHashMap<>();
c.put("ant", 3);
c.put("bee", 12);
c.put("cat", 8);
Integer maxEven = c.reduce(1,
    (k, v) -> v % 2 == 0 ? v : null, Integer::max); // 12 (3 skipped)
String big = c.search(1, (k, v) -> v > 10 ? k : null); // bee
System.out.println(maxEven + " " + big);            // 12 bee

The reducer should be associative and side-effect free, because the elements are combined in no fixed order and possibly on several threads. These methods are not locked snapshots: if other threads update the map during the call, the result reflects some of those changes.

ConcurrentSkipListMap and ConcurrentSkipListSet are the concurrent counterparts of TreeMap and TreeSet: they keep keys sorted by natural ordering or by a Comparator passed to the constructor, implement NavigableMap / NavigableSet (firstKey, headMap, ceiling, descendingSet, …), and are safe for many threads without locking the whole structure.

NavigableSet<Integer> s =
    new ConcurrentSkipListSet<>(List.of(40, 10, 30, 10));
System.out.println(s);                     // [10, 30, 40]
System.out.println(s.ceiling(25) + " " + s.floor(25)); // 30 10

NavigableMap<Integer, String> rooms = new ConcurrentSkipListMap<>();
rooms.put(310, "C");
rooms.put(101, "A");
rooms.put(205, "B");
System.out.println(rooms.tailMap(200));    // {205=B, 310=C}

They follow the same rules as their tree cousins: duplicates in a set are dropped, a Comparator passed to the constructor (such as Comparator.reverseOrder()) replaces natural ordering, first() means first in the set's own ordering, and elements must be mutually comparable — adding an object that does not implement Comparable to a set with no comparator throws ClassCastException. They follow the concurrent rules too: nulls are rejected with NullPointerException, and iterators and range views (headMap, subSet, …) are weakly consistent, so one thread can traverse a sorted range while others update the map, with no lock and no ConcurrentModificationException. A Collections.synchronizedNavigableMap(new TreeMap<>()) is also sorted and thread-safe per call, but iterating it or its views still requires holding its lock.

CopyOnWriteArrayList is a thread-safe List that copies its whole internal array on every modification (add, set, remove), so readers never lock and every iterator works on the array that existed when it was created. That makes reads and iteration very cheap and writes expensive: it is built for lists that are read and iterated far more often than they change, such as listener lists. CopyOnWriteArraySet is a Set built on the same idea (unsorted, backed by a CopyOnWriteArrayList).

Snapshot iteration has four consequences:

• Modifying the list during a loop is allowed and never throws ConcurrentModificationException.
• The loop does not see the changes: it visits exactly the elements of its snapshot, so adding inside the loop cannot make it run forever.
• An iterator created before a write does not see that write, even if it starts reading afterwards.
• The iterator cannot change the list: Iterator.remove, set and add throw UnsupportedOperationException. Change the list itself instead (list.remove(x), list.removeIf(...)).

List<String> list =
    new CopyOnWriteArrayList<>(List.of("p", "q", "r"));
for (String s : list) {
    if (s.equals("p")) {
        list.remove("r");   // allowed; the snapshot still has r
    }
    System.out.print(s + " ");
}
System.out.println(list);   // p q r [p, q]

list.listIterator().add("z"); // UnsupportedOperationException

Spliterators and streams use the same snapshot, so a stream pipeline that writes to its own CopyOnWriteArrayList source processes only the elements that were there when the terminal operation started, and the writes simply accumulate in the list. The same pipeline over an ArrayList source fails fast with ConcurrentModificationException. CopyOnWriteArrayList also offers addIfAbsent(e) and addAllAbsent(c), which check and add in one atomic step.

A BlockingQueue is a thread-safe queue whose extra methods wait: put waits for space and take waits for an element, which is exactly what a producer-consumer design needs. Every queue operation comes in four styles, and the style decides what happens when the queue is full (insert) or empty (remove or examine).

The waiting methods put, take and the timed offer/poll declare the checked InterruptedException, so code that calls them must catch it or declare it, or it does not compile. The non-blocking methods do not. A BlockingQueue also rejects null elements with NullPointerException, because null is the "nothing there" return value of poll.

BlockingQueue<String> q = new ArrayBlockingQueue<>(1);
System.out.println(q.offer("a") + " " + q.offer("b")); // true false
q.add("c");             // IllegalStateException: Queue full

Capacity depends on the implementation and its constructor:

• ArrayBlockingQueue — always bounded; the capacity is a required constructor argument.
• LinkedBlockingQueue — optionally bounded; the no-argument constructor gives a capacity of Integer.MAX_VALUE, so in practice put never waits.
• SynchronousQueue — no capacity at all: each put waits until another thread takes that element, and offer returns false unless a consumer is already waiting.
• PriorityBlockingQueue — unbounded, ordered by priority; LinkedBlockingDeque — a blocking double-ended queue.

ConcurrentLinkedQueue is thread-safe too, but it is not a BlockingQueue: it has no put or take, so a consumer can only poll and loop when it gets null.

Pick a concurrent collection by the access pattern the scenario states — how often it is written, whether readers must wait, whether order or indexing matters, and whether iteration may lock. The stated constraints usually rule out every type but one.

Watch the write rate: CopyOnWriteArrayList and CopyOnWriteArraySet are wrong for data that changes constantly, however attractive their snapshot iterators are. And watch the bound: a bounded queue with put is how a design makes producers wait; an unbounded one is how it guarantees they never do.

A parallel stream is a stream whose pipeline may be split into parts that run on several threads at once (by default in the common ForkJoinPool), with the partial results combined at the end. You get one in two ways: collection.parallelStream(), or parallel() on any existing stream, for example list.stream().parallel() or IntStream.range(0, n).parallel().

List<String> orders = List.of("o1", "o2");
System.out.println(orders.parallelStream().isParallel()); // true

Stream<Integer> s = orders.stream()
    .parallel()
    .map(String::length)
    .sequential();             // the last call wins
System.out.println(s.isParallel());                      // false

• The mode belongs to the whole pipeline, not to a stage. parallel() and sequential() just set a flag, and the last call before the terminal operation decides how every stage runs. You cannot run map in parallel and filter sequentially in one pipeline.
• isParallel() reports the current flag; it is itself an ordinary method, not a terminal operation.
• unordered() does not make a stream parallel; it only removes the encounter-order constraint (see below).
• Passing a collection to Stream.of wraps the collection itself as a single element (a Stream<List<T>>); a stream of the collection's elements comes from stream() or parallelStream().

Parallelism only pays off for large data and CPU-heavy, independent per-element work on a source that splits well (arrays, ArrayList, ranges). For small inputs the splitting and merging cost more than they save, and blocking I/O inside a parallel stream ties up the shared pool's threads.

A parallel stream keeps the encounter order of an ordered source in its results, but not in the timing of its side effects. Lists, arrays, Stream.of, ranges and sorted sets are ordered; HashSet, Set.of and HashMap views are not. The rule, operation by operation:

String j = List.of("k", "l", "m").parallelStream()
    .peek(System.out::print)          // letters in any order
    .map(String::toUpperCase)
    .collect(Collectors.joining(","));
System.out.println(" " + j);          // ... K,L,M

IntStream.rangeClosed(10, 30).parallel()
    .filter(n -> n % 4 == 0)          // 12 16 20 24 28
    .skip(2).limit(2)
    .boxed().toList();                // [20, 24]

Three traps follow from the table. First, nothing earlier in the pipeline makes forEach ordered: an ordered source or an intermediate sorted fixes the encounter order, but a parallel forEach ignores it; use forEachOrdered or collect the result. Second, a stream that is unordered gives no order to preserve: Set.of(3, 1, 2).parallelStream().toList() has an unspecified order, and so does any stream after unordered(), where limit and distinct may also pick different elements. Third, peek is for debugging only: the implementation may skip it entirely when the terminal operation can compute its result without visiting the elements, so never rely on it to produce output or state.

The safe way to gather a parallel stream's results is a collector or a reduction, not a lambda that adds to a shared collection. A lambda like forEach(out::add) runs on many threads at once; if the target is any plain, non-thread-safe collection, updates are lost, or an exception such as ArrayIndexOutOfBoundsException is thrown, and the result can differ on every run. The capture compiles fine — the variable is effectively final; it is the object that is unsafe. Only a genuinely thread-safe type helps; a wrapper that adds some other check does not.

List<String> names = new ArrayList<>();
words.parallelStream()
    .map(String::trim)
    .forEach(names::add);         // may lose elements or throw

List<String> safe = words.parallelStream()
    .map(String::trim)
    .toList();                    // every element, in order

If you truly need a shared target, make it thread-safe: a CopyOnWriteArrayList, a synchronized wrapper or a ConcurrentLinkedQueue keeps every element (in no particular order), and a ConcurrentHashMap updated with merge keeps every count. But a collector is cleaner and usually faster.

Concurrent collectors

Ordinary toMap and groupingBy build a separate map for each chunk of the split stream and merge them, which is expensive for maps. The concurrent versions have all threads insert into one shared ConcurrentMap:

• Collectors.toConcurrentMap(keyFn, valueFn) returns a ConcurrentMap; like toMap, a duplicate key throws IllegalStateException unless you pass a merge function as the third argument to decide which value wins.
• Collectors.groupingByConcurrent(classifier) returns a ConcurrentMap<K, List<T>>. It is a CONCURRENT and UNORDERED collector, so the elements inside each group's list can be in any order, even from an ordered source; use groupingBy when the order within groups matters.

ConcurrentMap<Integer, String> first =
    Stream.of("ant", "bee", "wasp").parallel()
        .collect(Collectors.toConcurrentMap(
            String::length, w -> w,
            (x, y) -> x.compareTo(y) < 0 ? x : y));
System.out.println(first.get(3) + " " + first.get(4)); // ant wasp

Because the merge function above picks the alphabetically smaller word, the result does not depend on which thread got there first. Neither of these maps has a predictable iteration order, so look values up by key or copy the map into a sorted map when you need ordered output. The rules for reduce identities and combiners, collect(supplier, accumulator, combiner) and gatherers such as mapConcurrent are covered in the stream reduction lesson (objective 6.2).