Mayank Pant
Go back

Concurrent Collection

Table of Contents

Open Table of Contents

Problem with traditional Collection

In Traditional Collection object we already have both thread safe and non-thread safe objects.

Non-thread safe classes are: ArrayList, LinkedList, HashMap, etc,. Problem with Traditional non-thread safe Collection classes: It can be accessed by multiple threads simultaneously and there may be a chance of data inconsistency problem.

We can also make the traditional Collection thread safe by using the utility methods provided by Collections utility class. Example:

Thread safe classes are: Stack, Vector, HashTable, etc,.

Problem with traditional thread-safe Collection, and Collection classes made thread safe using Collections utility class:

Example:

List<String> list = new ArrayList<>();
list.add("One");
list.add("Two");
list.add("Three");

Thread readerThread = new Thread(() -> {
	System.out.println("Reader thread started);
	for(String str : list) {
		System.out.println("Reading: " + str);
	}
});

Thread writerThread = new Thread(() -> {
	System.out.println("Modify thread started");
	list.add("Five");
});

readerThread.start();
writerThread.start();

//Output:
//ConcurrentModificationException: only when reader is working and writer tries to update the value
//If writer would have been started first then this issue might not have come

Problems with Traditional Approaches:

To overcome this problem, Concurrent Collection has been introduced in Java 1.5 version.

Concurrent Collections Benefits:

To solve the above code problem using Concurrent Collection:

List<String> list = new CopyOnWriteArrayList<>();
//..
//..
//Same code as above
//..
//..
readerThread.start();
writerThread.start();

//wait for threads to complete execution
readerThread.join();
writerThread.join();

//Now check if modification is done or not
System.out.println(list);

Need for Concurrent Collections

In multithreaded environment multiple threads will write and read shared data, this may lead to data inconsistency or data corruption.

So Concurrent Collection helps prevent data inconsistency or data corruption in multi-threaded environment. And it allows multiple threads to work simultaneously without creating any issues.

Example:

Understanding Concurrent Collection

Important concurrent classes/interfaces are:

Four distinct tools for four distinct problems.

concurrent-collection

ConcurrentHashMap

The ConcurrentHashMap is very similar to the java.util.HashTable class, except that ConcurrentHashMap offers better concurrency than HashTable does.

ConcurrentHashMap does not lock the Map while you are reading from it. Additionally, ConcurrentHashMap does not lock the entire Map when writing to it. It only locks the part of the Map that is being written to, internally. It allows read operations concurrently and update operations in a thread safe manner

Features of ConcurrentHashMap:

  1. ConcurrentHashMap internally uses HashTable as its data structure. It is thread safe just like Hashtable. It provides all the functionalities of HashMap except thread safety.
  2. ConcurrentHashMap internally divides it into segments. Each segment works independently and can be accessed by different reader threads simultaneously. However, each segment can be accessed only by one writer thread at a time. This also means, a concurrent hash map can be accessed by as many writer threads together as there are segments.
  3. The default level of concurrency is 16. Which means by default, there are 16 segments.
  4. Read operations don’t require locking of concurrent hash map, where as write operations do require locking.
  5. Locking is known as segment or bucket locking.
  6. Concurrent hash map doesn’t allow null key or null values.

Another difference is that ConcurrentHashMap does not throw ConcurrentModificationException if the ConcurrentHashMap is changed while being iterated.

hashmap-segment

ConcurrentHashMap class hierarchy

The java.util.concurrent.ConcurrentMap interface represents a Java Map which is capable of handling concurrent access (puts and gets) to it.

The ConcurrentMap has a few extra atomic methods in addition to the methods it inherits from its superinterface, java.util.Map.

Since ConcurrentMap is an interface, you need to use one of its implementations in order to use it. The java.util.concurrent package contains the following implementations of the ConcurrentMap interface:

concurrent-class-hierarchy

ConcurrentHashMap operations

Here is an example of how to use the ConcurrentMap interface. The example uses a ConcurrentHashMap implementation:

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

// ── Basic operations (same as HashMap, fully thread-safe) ───────── 
map.put("alice", 1); 
map.get("alice"); // non-blocking — no lock at all 
map.remove("alice"); 
map.containsKey("bob"); 

// ── Atomic conditional operations ───────────────────────────────── 
map.putIfAbsent("alice", 1); // only inserts if key absent — atomic. Returns null if succeeds, else returns old value if key already present
map.replace("alice", 1, 2); // only replaces if the given key's current value is 1 
map.remove("alice", 1); // only removes if given key's value equals 1 


// ── compute family — atomic read-modify-write ───────────────────── 
// Only acquires a lock if the method condition -- ifAbsent, ifPresent -- succeeds. After acquiring lock, executes the given lambda to update the value

// computeIfAbsent: create + insert ONLY if key missing 
// Classic use: lazily build per-key collections 
map.computeIfAbsent("alice", k -> 0); // inserts 0 only if absent 
List<String> list = perUserMap.computeIfAbsent("alice", k -> new ArrayList<>()); 

// computeIfPresent: update ONLY if key exists 
map.computeIfPresent("alice", (k, v) -> v + 1); // increment if present 

// compute: always called — create if absent OR update if present 
map.compute("alice", (k, v) -> v == null ? 1 : v + 1); // init or increment 

// merge: simplest counting/accumulation pattern 
map.merge("alice", 1, Integer::sum); // set 1 if absent; else sum existing + 1 
// ← THE cleanest word-count / frequency-count idiom 

// ── Parallel bulk operations (Java 8+) ──────────────────────────── 
// threshold: min entry count per parallel task (1 = max parallelism) 
map.forEach(1, (k, v) -> process(k, v)); 
long totalScore = map.reduceValues(1, v -> (long) v, Long::sum); 
String firstHigh = map.search(1, (k, v) -> v > 100 ? k : null); 

// ── Size caveat ─────────────────────────────────────────────────── 
int approx = map.size(); // may miss in-flight concurrent updates 
long better = map.mappingCount(); // returns long — better for large maps


//Runnint concurrent reader and writer threads on hashmap
Thread readerThread = new Thread(() -> {
	System.out.println("Reader thread working:");
	for(String key : map.keySet()) {
		System.out.println("Key: " + key + ", Value: " + map.get(key));
	}
});
readerThread.start();

Thread writerThread = new Thread(() -> {
	System.out.println("Writer Thread working");
	map.put("Ten", 10);
});
writerThread.start();
//No error

//If we change ConcurrentHashMap with HashMap, then it will throw ConcurrentModificaitonException

Fail-safe vs fail-fast

ConcurrentHashMap returns a fail safe iterator. It means, using this iterator, we can modify ConcurrentHashMap while iterating it. Let’s see one example:

public class ConcurrentHashMapExample {
	
	public static void main(String[] args) {
		
		ConcurrentHashMap<Integer, String> cMap = new ConcurrentHashMap<>();
		cMap.put(1, "Taj Mahal");
		cMap.put(2, "Qutab Minar");
		cMap.put(3, "Char Minar");
		
		Iterator<Integer> it = cMap.keySet().iterator();
		while(it.hasNext()) {
			int key = it.next();
			if(key == 2)
				cMap.put(2, "Gateway of India");
			System.out.println(key + " : " + cMap.get(key));
		}
	}
}

In the above example, we saw that we can modify ConcurrentHashMap while iterating it. It became possible because the iterator returned by ConcurrentHashMap is fail safe. Which means we can modify it while iterating.

Avoiding Slipped Conditions

Look at this Java code example:

ConcurrentMap map = new ConcurrentHashMap();

if( !map.containsKey("key1") ) {
     map.put("key1", "value1");
}

Even if both the containsKey() and put() method are both thread safe - the above if-statement construct is not thread safe.

The problem with the above if-construct is that if 2 threads execute the above if-statement simultaneously they may both call map.containsKey("key1") at the same time, they might both receive the answer that the map does not contain the key “key1” - meaning containsKey() returns false. In that case, both threads will continue into the if-statement body, and both insert a value into the ConcurrentMap. The second thread to insert its value will overwrite the value inserted by the first thread.

In the above example, having 2 threads both insert the static value “value1” may not be so big of a problem, but what if the value was computed - based on something the individual thread knows?

The solution to this problem is to use one of the atomic methods putIfAbsent() or computeIfAbsent() instead.

The putIfAbsent() method of the Java ConcurrentMap interface inserts the given key + value pair, if no key + value pair exists for the given key already. The ConcurrentMap implementation will make sure that only one thread at a time will be allowed to insert a value for the same key. The ConcurrentMap might allow multiple threads to insert key + value pairs for different keys - depending on its internal implementation (e.g. some implementations may only allow multiple concurrent insertions if the key + value pair lands in different “buckets” internally).

The computeIfAbsent() method is similar to the putIfAbsent() method - except it enables you to compute a value to insert for a given key, if no key + value pair is already stored for that key. Only one thread at a time will be allowed to execute computeIfAbsent() for the same key. If two threads calls computeIfAbsent() for the same absent key, one of the threads will be allowed to compute and insert its value, and the other thread will not compute nor insert any value.

The value to be inserted is computed by the Java lambda expression that you pass in as the second parameter to the computeIfAbsent() method.

HashMap vs ConcurrentHashMap

hashpmap-vs-concurrenthashmap

Key notes on ConcurrentHashMap

CopyOnWriteArrayList

On its working: The CopyOnWriteArrayList is a thread safe version of ArrayList. If we are making modifications like adding, removing elements in CopyOnWriteArrayList, then JVM does so by creating a new copy of it by the use of Cloning. Every mutation on CopyOnWriteArrayList (add, remove, set) acquires a lock, copies the entire backing array, applies the change to the copy, then atomically replaces the reference. We can also add duplicate elements in it.

On Reading: Multiple threads can read the data from CopyOnWriteArrayList, but only one thread can write data at a particular time. Reads and iterations hold a reference to the snapshot at the time they started — they never block and never see a ConcurrentModificationException.

The tradeoff: The tradeoff: writes are O(N) and iterators may be stale.

When can it be used: CopyOnWriteArrayList is costly if used in case of more update operations. Because when changes are made, JVM has to create a cloned copy of the underlying array and add/update elements to it. CopyOnWriteArrayList is the best choice in multithreading, if there are more read operations.

Code Example: Code is similar to that of ArrayList and we have already seen it in the “Problems with traditional Collection” section.

Class hierarchy: copy-on-write-arraylist-class-hierarchy

Copy-on-write mechanism, safe iteration, and the staleness tradeoff:

CopyOnWriteArrayList<EventListener> listeners = new CopyOnWriteArrayList<>(); 

// ── Writes: lock + copy entire array + mutate + atomic swap ─────── 
listeners.add(newListener); // copies [A,B] → writes [A,B,C] — O(N) 
listeners.remove(oldListener); // copies [A,B,C] → writes [A,C] — O(N) 
listeners.addIfAbsent(listener); // atomic: add only if not already present 
listeners.set(0, updatedListener); // replaces element — still O(N) copy 

// ── Reads: non-blocking, zero locking ───────────────────────────── 
EventListener first = listeners.get(0); // reads current snapshot, no lock 
int count = listeners.size(); // reads current snapshot 

// ── Iteration: safe snapshot — NEVER throws ConcurrentModificationException 
// Iterator captures a reference to the array at iterator creation time 
// New elements added AFTER the iterator was created are NOT visible 
// This is correct and expected — iteration always completes safely 
for (EventListener l : listeners) { 
	l.onEvent(event); // safe — another thread may add/remove without affecting this loop 
} 

// Explicit snapshot — fire-and-forget event dispatch pattern: 
listeners.forEach(l -> l.onEvent(event)); // iterates snapshot, fully concurrent-safe 

// ── Stale iterator — by design: ─────────────────────────────────── 
Iterator<EventListener> ite = listeners.iterator(); // snapshot: [A, B, C] 
listeners.add(newListener); // list is now [A, B, C, D] 
// ite iterator still sees [A, B, C] — newListener not visible in this iteration 

// ── CopyOnWriteArraySet: same concept for Set semantics ─────────── 
CopyOnWriteArraySet<String> cowSet = new CopyOnWriteArraySet<>(); 
cowSet.add("handler-1"); // copies the array, enforces uniqueness 

// ── Use when / avoid when ───────────────────────────────────────── 
// ✓ Use: event listener lists, observer registries, callback registries 
// → reads (dispatch) vastly outnumber writes (register/unregister) 
// ✗ Avoid: large collections, frequent writes, must see latest data 
// ✗ Avoid: hot loop reads — creating iterators on every tight loop iteration

Key notes:

CopyOnWriteArraySet

Code: Just update the first line in the CopyOnWriteArrayList code to Set<String> set = new CopyOnWriteArraySet<>();

Class hierarchy: copy-on-write-arrayset

BlockingQueue

BlockingQueue is the backbone of producer-consumer patterns.

The Java BlockingQueue interface, java.util.concurrent.BlockingQueue, represents a queue which is thread safe to put elements into, and take elements out of from. In other words, multiple threads can be inserting and taking elements concurrently from a Java BlockingQueue, without any concurrency issues arising.

The term blocking queue comes from the fact that the Java BlockingQueue is capable of blocking the threads that try to insert or take elements from the queue. For instance, if a thread tries to take an element and there are none left in the queue, the thread can be blocked until there is an element to take. Whether or not the calling thread is blocked depends on what methods you call on the BlockingQueue.

Since BlockingQueue is an interface, you need to use one of its implementations to use it. Each variant makes different tradeoffs between throughput, fairness, ordering, and memory. The java.util.concurrent package has the following implementations of the BlockingQueue interface:

The four-method families give you blocking, non-blocking, time-bounded, and exception-throwing variants for each direction.

LinkedBlockingQueue typically has higher throughput than ArrayBlockingQueue because its separate head/tail locks let producers and consumers operate independently.

BlockingQueue variants — ArrayBQ, LinkedBQ, SynchronousQueue, PriorityBQ:

// ── Four method families: same operation, different failure modes ── 
//                Put (add to tail)    Take (remove from head) 
// Blocks:        put(e)               take() 
// Non-blocking:  offer(e) → bool      poll() → E | null 
// Timed:         offer(e, t, unit)    poll(t, unit) → E | null 
// Throws:        add(e) throws ISE    remove() throws NSEE 

// ── ArrayBlockingQueue: bounded, single lock, fair option ───────── 
BlockingQueue<Task> aq = new ArrayBlockingQueue<>(100); // capacity 100 
BlockingQueue<Task> fair= new ArrayBlockingQueue<>(100, true); // FIFO fairness 

// ── LinkedBlockingQueue: optionally bounded, TWO locks ──────────── 
// Separate takeLock (head) and putLock (tail) → producers + consumers 
// don't contend with each other at all → higher throughput than ArrayBQ 
BlockingQueue<Task> lq = new LinkedBlockingQueue<>(); // unbounded ⚠ 
BlockingQueue<Task> lb = new LinkedBlockingQueue<>(1000); // bounded ✓ 

// ── SynchronousQueue: zero-capacity, direct hand-off ────────────── 
// put() BLOCKS until a consumer is ready to take() — no buffering at all 
// Producer and consumer must rendezvous — perfect for direct task handoff 
BlockingQueue<Task> sq = new SynchronousQueue<>(); 
// Used internally by Executors.newCachedThreadPool() for direct task handoff 

// ── PriorityBlockingQueue: unbounded, priority-ordered ──────────── 
// take() always returns the highest-priority element (min heap internally) 
BlockingQueue<Task> pq = new PriorityBlockingQueue<>(11, 
	Comparator.comparingInt(Task::getPriority)); 
	
// ── Producer-Consumer thread-pool worker pattern ────────────────── 
BlockingQueue<Task> workQ = new LinkedBlockingQueue<>(500); // bounded! 

// Producer (submitter thread): 
while (!done) { 
	workQ.put(generateTask()); // blocks if 500 tasks queued — natural backpressure 
} 
workQ.put(POISON_PILL); // signal consumers to stop 

// Consumer (worker thread): 
while (true) { 
	Task task = workQ.take(); // blocks when empty — zero CPU waste 
	if (task == POISON_PILL) break; 
	processTask(task); 
}

Key notes:

ArrayBlockingQueue

ArrayBlockingQueue

LinkedBlockingQueue

LinkedBlockingQueue

SynchronousQueue

SynchronousQueue

PriorityBlockingQueue

PriorityBlockingQueue

DelayQueue

DelayQueue

ConcurrentLinkedQueue

ConcurrentLinkedQueue does NOT implement BlockingQueue

ConcurrentLinkedQueue is a non-blocking, thread-safe queue that allows multiple threads to add and remove elements simultaneously without waiting, while LinkedBlockingQueue is a blocking queue that can cause threads to wait when the queue is empty or full. The choice between them depends on whether you need blocking behavior or higher throughput without waiting.

ConcurrentLinkedQueue vs LinkedBlockingQueue

While both LinkedBlockingQueue and ConcurrentLinkedQueue are thread-safe, they operate differently:

The Importance of Blocking and Non-Blocking Queues

Understanding blocking and non-blocking queues enhances your ability to design efficient, thread-safe systems. Here’s why this knowledge is vital:

Understanding ConcurrentLinkedQueue

ConcurrentLinkedQueue uses the Michael-Scott non-blocking queue algorithm. offer() and poll() perform CAS on the tail and head nodes respectively — if the CAS fails (another thread beat them), they retry. No thread ever blocks or holds a lock. This makes it the highest-throughput option under extreme concurrency, but it requires the caller to handle the null-on-empty behavior rather than blocking.

ConcurrentLinkedQueue — lock-free operations, draining, and pitfalls:

Queue<Task> clq = new ConcurrentLinkedQueue<>(); 

// ── Core non-blocking operations ────────────────────────────────── 
clq.offer(task); // always true (unbounded) — lock-free CAS on tail 
Task t = clq.poll(); // null if empty — lock-free CAS on head 
Task t = clq.peek(); // returns head WITHOUT removing — lock-free 
boolean empty = clq.isEmpty(); // O(1), reliable 
int n = clq.size(); // ⚠ O(N) — traverses entire list! never in hot path 

// ── High-throughput multi-producer pattern ───────────────────────── 
// Thousands of threads offering concurrently — CAS retries on contention 
// Under low contention: O(1) with no spin. Under high contention: brief spin. 
IntStream.range(0, 10_000).parallel() 
	.forEach(i -> clq.offer(new Task(i))); // all concurrent, no locks 
	
// ── Draining in a consumer loop ──────────────────────────────────── 
Task task; 
while ((task = clq.poll()) != null) { 
	process(task); // stops when queue is empty (poll returns null) 
} 

// ── Draining a batch ────────────────────────────────────────────── 
List<Task> batch = new ArrayList<>(100); 
Task item; 
while (batch.size() < 100 && (item = clq.poll()) != null) { 
	batch.add(item); 
} 
// Note: ConcurrentLinkedQueue does NOT implement BlockingQueue 
//       → no drainTo(), no put(), no take() 

// ── ConcurrentLinkedDeque: double-ended version ─────────────────── 
Deque<Task> cld = new ConcurrentLinkedDeque<>(); 
cld.offerFirst(urgent); // push to head — high-priority insertion 
cld.offerLast(normal); // append to tail — normal insertion 
cld.pollFirst(); // remove from head — LIFO or FIFO depending on usage 
cld.pollLast(); // remove from tail 

// ── CLQ vs BlockingQueue — when to choose each ──────────────────── 
// ConcurrentLinkedQueue: 
// ✓ CPU-bound consumer that never needs to wait 
// ✓ Multiple producers + single fast consumer 
// ✓ Unbounded is acceptable (or managed externally) 
// BlockingQueue: 
// ✓ Consumer should sleep when empty (not busy-wait) 
// ✓ Need backpressure (bounded put()) 
// ✓ Producer-consumer with mismatched rates

Key notes:

BlockingDeque

BlockingDeque

LinkedBlockingDeque

LinkedBlockingDeque

Decision Guide: which to use when

Choosing the wrong concurrent collection is one of the most common performance bugs. The decision comes down to four questions: Do you need blocking semantics? How read-heavy is it? Do you need ordering guarantees? Is memory bounded important? This guide maps each scenario to the right tool.

Decision guide — matching each collection to its use case:

// ═══════════════════════════════════════════════════════════════════ 
// DECISION TREE 
// ═══════════════════════════════════════════════════════════════════ 
// 
// Need a Map? 
// → ConcurrentHashMap — always. Replaced synchronizedMap() everywhere. 
//     ✓ compute/merge for atomic read-modify-write 
//     ✗ Don't: Collections.synchronizedMap(new HashMap<>()) 
// 
// Need a List — reads >> writes (listener lists, observer registries)? 
// → CopyOnWriteArrayList // ✓ Iteration always safe, zero lock overhead on reads 
//     ✗ Avoid if writes are frequent or list is large (> ~100 elements) 
// 
// Need a Queue — producer/consumer with backpressure and blocking? 
// → BlockingQueue (pick a variant): 
//     ArrayBlockingQueue: bounded, predictable memory 
//     LinkedBlockingQueue: bounded (always specify!), higher throughput 
//     PriorityBlockingQueue: ordered by priority, unbounded 
//     SynchronousQueue: direct handoff, zero buffering 
// 
// Need a Queue — non-blocking, high-throughput, caller handles null? 
// → ConcurrentLinkedQueue 
//     ✓ Lock-free, excellent under many producers 
//     ✗ Not a replacement for BlockingQueue if consumer must sleep 
// 
// Need a Set? 
// → ConcurrentHashMap.newKeySet() or Collections.newSetFromMap(new CHM<>()) 
// → CopyOnWriteArraySet (for small, read-heavy sets) 

// ── Common patterns ──────────────────────────────────────────────── 

// Word frequency count (CHM + merge) 
ConcurrentHashMap<String, Integer> freq = new ConcurrentHashMap<>(); 
words.parallelStream().forEach(w -> freq.merge(w, 1, Integer::sum)); 

// Thread-safe Set (CHM-backed) 
Set<String> concurrentSet = ConcurrentHashMap.newKeySet(); 
concurrentSet.add("alice"); 

// Worker pool with bounded queue + backpressure 
BlockingQueue<Runnable> tasks = new LinkedBlockingQueue<>(1000); 
ThreadPoolExecutor pool = new ThreadPoolExecutor( 
	4, 8, 60, TimeUnit.SECONDS, tasks, 
	new ThreadPoolExecutor.CallerRunsPolicy() // backpressure on full queue 
); 

// High-frequency event bus (CLQ + drain batch) 
ConcurrentLinkedQueue<Event> events = new ConcurrentLinkedQueue<>(); 
// Many producer threads: events.offer(e) — lock-free, fast 
// Single background thread: drain and process in batches 

// ── Antipatterns to avoid ────────────────────────────────────────── 
// ✗ Collections.synchronizedList(new ArrayList<>()) — global lock, slow 
// ✗ Collections.synchronizedMap(new HashMap<>()) — replaced by CHM 
// ✗ new LinkedBlockingQueue<>() without capacity — unbounded, OOM risk 
// ✗ clq.size() in a tight loop — O(N) per call!

Key notes:

The ConcurrentHashMap compute family — why it matters

The entire compute family solves the check-then-act problem atomically. Without it, even with a ConcurrentHashMap, this is a race:

// ✗ Race condition — check and act are two separate steps:
if (!map.containsKey("alice")) {    // T1 checks: absent
    map.put("alice", 1);            // T2 also checks: absent — T1 and T2 both put 1!
}

// ✓ Atomic — check and insert happen in one bin-locked operation:
map.computeIfAbsent("alice", k -> 1);
map.merge("alice", 1, Integer::sum);  // cleanest counter pattern

The function you pass to compute/merge/computeIfAbsent runs inside the bin lock — so it’s guaranteed to see a consistent view and no other thread can modify that key concurrently. The consequence: keep these functions fast and non-blocking, or you’ll serialize access to that bin.

The blocked thread cost comparison

ApproachThread state while waitingCPU usage
BlockingQueue.take()WAITING — thread suspended0%
ConcurrentLinkedQueue.poll() retry loopRUNNABLE — busy spin100% per core
synchronized collectionBLOCKED — waiting for monitor0%
ConcurrentHashMap.get()RUNNABLE — no wait at allminimal

BlockingQueue is almost always the right choice for producer-consumer because the consumer thread sleeps when there’s nothing to do — it doesn’t waste a CPU core. ConcurrentLinkedQueue requires the application to decide what to do on a null poll, which typically means either busy-waiting (CPU waste) or some external signalling mechanism.

Best Practices

Common Patterns:

Resources

Share this post on:
Share this post via WhatsApp Share this post on Facebook Share this post on X Share this post via Telegram Share this post on Pinterest Share this post via email
Next Post
Generics in Java