Concurrent collections in Java are thread-safe implementations of collection interfaces designed for multi-threaded environments. They provide better performance than synchronized wrappers by using advanced synchronization techniques.
1. ConcurrentMap Implementations
ConcurrentHashMap
import java.util.concurrent.*;
import java.util.*;
public class ConcurrentHashMapExample {
public static void main(String[] args) {
// Creating ConcurrentHashMap
ConcurrentHashMap<String, Integer> concurrentMap = new ConcurrentHashMap<>();
// Basic operations - thread-safe
concurrentMap.put("John", 25);
concurrentMap.put("Alice", 30);
concurrentMap.putIfAbsent("John", 40); // Won't replace existing
concurrentMap.putIfAbsent("Bob", 35); // Will be added
System.out.println("Map: " + concurrentMap);
// Atomic operations
concurrentMap.compute("John", (k, v) -> v + 1); // John becomes 26
concurrentMap.merge("Alice", 1, Integer::sum); // Alice becomes 31
System.out.println("After atomic operations: " + concurrentMap);
// Search operations
String result = concurrentMap.search(1, (k, v) -> v > 30 ? k : null);
System.out.println("First person older than 30: " + result);
// ForEach operations
concurrentMap.forEach(1, (k, v) -> System.out.println(k + ": " + v));
// Reduce operations
int totalAge = concurrentMap.reduceValues(1, Integer::sum);
System.out.println("Total age: " + totalAge);
// Working with ConcurrentHashMap in multi-threaded environment
multiThreadedExample();
}
public static void multiThreadedExample() {
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
ExecutorService executor = Executors.newFixedThreadPool(3);
// Multiple threads writing concurrently
for (int i = 0; i < 10; i++) {
final int threadId = i;
executor.submit(() -> {
for (int j = 0; j < 100; j++) {
String key = "key-" + threadId + "-" + j;
map.put(key, threadId * 100 + j);
}
});
}
executor.shutdown();
try {
executor.awaitTermination(1, TimeUnit.MINUTES);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Final map size: " + map.size());
}
}
ConcurrentHashMap Advanced Features
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
public class ConcurrentHashMapAdvanced {
public static void main(String[] args) {
ConcurrentHashMap<String, AtomicInteger> wordCount = new ConcurrentHashMap<>();
// Thread-safe accumulation
List<String> words = Arrays.asList("apple", "banana", "apple", "orange", "banana", "apple");
words.parallelStream().forEach(word -> {
wordCount.computeIfAbsent(word, k -> new AtomicInteger(0)).incrementAndGet();
});
System.out.println("Word counts: " + wordCount);
// Bulk operations
ConcurrentHashMap<String, String> config = new ConcurrentHashMap<>();
config.put("host", "localhost");
config.put("port", "8080");
config.put("timeout", "30");
// Replace all values
config.replaceAll((k, v) -> k.toUpperCase() + "_" + v);
System.out.println("Transformed config: " + config);
// Compute if absent/present
ConcurrentHashMap<String, List<String>> userPreferences = new ConcurrentHashMap<>();
List<String> preferences = userPreferences.computeIfAbsent("user1",
k -> new CopyOnWriteArrayList<>());
preferences.add("theme:dark");
userPreferences.computeIfPresent("user1", (k, v) -> {
v.add("lang:en");
return v;
});
System.out.println("User preferences: " + userPreferences);
}
}
2. Concurrent Queues
Blocking Queues
import java.util.concurrent.*;
import java.util.*;
public class BlockingQueueExample {
public static void main(String[] args) throws InterruptedException {
// ArrayBlockingQueue - bounded queue
BlockingQueue<String> arrayQueue = new ArrayBlockingQueue<>(3);
// LinkedBlockingQueue - optionally bounded
BlockingQueue<Integer> linkedQueue = new LinkedBlockingQueue<>(5);
// PriorityBlockingQueue - unbounded priority queue
BlockingQueue<Task> priorityQueue = new PriorityBlockingQueue<>();
// SynchronousQueue - direct handoff
BlockingQueue<String> syncQueue = new SynchronousQueue<>();
demonstrateArrayBlockingQueue();
demonstrateProducerConsumer();
}
public static void demonstrateArrayBlockingQueue() throws InterruptedException {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
// Add elements
queue.put("Item1");
queue.put("Item2");
queue.put("Item3");
System.out.println("Queue: " + queue);
// Try to add more - will block until space available
new Thread(() -> {
try {
queue.put("Item4"); // This will block until space is available
System.out.println("Item4 added");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
Thread.sleep(1000);
// Remove an item to make space
String item = queue.take();
System.out.println("Removed: " + item);
}
public static void demonstrateProducerConsumer() {
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
// Producer
Thread producer = new Thread(() -> {
try {
for (int i = 0; i < 10; i++) {
queue.put(i);
System.out.println("Produced: " + i);
Thread.sleep(100);
}
queue.put(-1); // Poison pill
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
// Consumer
Thread consumer = new Thread(() -> {
try {
while (true) {
Integer item = queue.take();
if (item == -1) break; // Poison pill
System.out.println("Consumed: " + item);
Thread.sleep(200);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
}
}
class Task implements Comparable<Task> {
private String name;
private int priority;
public Task(String name, int priority) {
this.name = name;
this.priority = priority;
}
@Override
public int compareTo(Task other) {
return Integer.compare(other.priority, this.priority); // Higher priority first
}
@Override
public String toString() {
return name + "(" + priority + ")";
}
}
Non-Blocking Queues
import java.util.concurrent.*;
public class NonBlockingQueueExample {
public static void main(String[] args) {
// ConcurrentLinkedQueue - non-blocking FIFO
ConcurrentLinkedQueue<String> nonBlockingQueue = new ConcurrentLinkedQueue<>();
// ConcurrentLinkedDeque - non-blocking double-ended queue
ConcurrentLinkedDeque<String> deque = new ConcurrentLinkedDeque<>();
demonstrateConcurrentLinkedQueue();
}
public static void demonstrateConcurrentLinkedQueue() {
ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
// Multiple producers
Thread producer1 = new Thread(() -> {
for (int i = 0; i < 5; i++) {
queue.offer("Producer1-" + i);
}
});
Thread producer2 = new Thread(() -> {
for (int i = 0; i < 5; i++) {
queue.offer("Producer2-" + i);
}
});
// Single consumer
Thread consumer = new Thread(() -> {
int count = 0;
while (count < 10) {
String item = queue.poll();
if (item != null) {
System.out.println("Consumed: " + item);
count++;
}
}
});
producer1.start();
producer2.start();
consumer.start();
try {
producer1.join();
producer2.join();
consumer.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Final queue size: " + queue.size());
}
}
3. Concurrent Lists and Sets
CopyOnWrite Collections
import java.util.concurrent.*;
import java.util.*;
public class CopyOnWriteExample {
public static void main(String[] args) {
// CopyOnWriteArrayList - thread-safe list
CopyOnWriteArrayList<String> copyOnWriteList = new CopyOnWriteArrayList<>();
// CopyOnWriteArraySet - thread-safe set
CopyOnWriteArraySet<String> copyOnWriteSet = new CopyOnWriteArraySet<>();
demonstrateCopyOnWriteArrayList();
demonstrateCopyOnWriteIteration();
}
public static void demonstrateCopyOnWriteArrayList() {
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>(
Arrays.asList("A", "B", "C")
);
// Multiple threads reading and writing
Thread writer = new Thread(() -> {
for (int i = 0; i < 5; i++) {
String newItem = "New-" + i;
list.add(newItem);
System.out.println("Added: " + newItem);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
Thread reader = new Thread(() -> {
for (int i = 0; i < 3; i++) {
System.out.println("Current list: " + list);
try {
Thread.sleep(150);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
writer.start();
reader.start();
try {
writer.join();
reader.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public static void demonstrateCopyOnWriteIteration() {
CopyOnWriteArrayList<Integer> numbers = new CopyOnWriteArrayList<>(
Arrays.asList(1, 2, 3, 4, 5)
);
// Iterator uses snapshot of the list when created
Iterator<Integer> iterator = numbers.iterator();
// Modify list while iterating
numbers.add(6);
numbers.remove(Integer.valueOf(3));
// Iterator still sees original snapshot
System.out.println("Iterator sees:");
while (iterator.hasNext()) {
System.out.println(iterator.next()); // Won't see changes
}
System.out.println("Actual list: " + numbers);
}
}
ConcurrentSkipList Collections
import java.util.concurrent.*;
import java.util.*;
public class ConcurrentSkipListExample {
public static void main(String[] args) {
// ConcurrentSkipListSet - concurrent sorted set
ConcurrentSkipListSet<Integer> skipListSet = new ConcurrentSkipListSet<>();
// ConcurrentSkipListMap - concurrent sorted map
ConcurrentSkipListMap<String, Integer> skipListMap = new ConcurrentSkipListMap<>();
demonstrateConcurrentSkipListSet();
demonstrateConcurrentSkipListMap();
}
public static void demonstrateConcurrentSkipListSet() {
ConcurrentSkipListSet<Integer> set = new ConcurrentSkipListSet<>();
// Add elements from multiple threads
ExecutorService executor = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
final int start = i * 10;
executor.submit(() -> {
for (int j = 0; j < 10; j++) {
set.add(start + j);
}
});
}
executor.shutdown();
try {
executor.awaitTermination(1, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Sorted set: " + set);
System.out.println("First: " + set.first());
System.out.println("Last: " + set.last());
System.out.println("Subset [20-40]: " + set.subSet(20, 40));
}
public static void demonstrateConcurrentSkipListMap() {
ConcurrentSkipListMap<String, Employee> employeeMap = new ConcurrentSkipListMap<>();
employeeMap.put("John", new Employee("John", 50000));
employeeMap.put("Alice", new Employee("Alice", 60000));
employeeMap.put("Bob", new Employee("Bob", 55000));
employeeMap.put("Carol", new Employee("Carol", 70000));
System.out.println("Employee map: " + employeeMap);
// Navigable map operations
System.out.println("First entry: " + employeeMap.firstEntry());
System.out.println("Last entry: " + employeeMap.lastEntry());
System.out.println("Ceiling key for 'C': " + employeeMap.ceilingKey("C"));
System.out.println("Floor key for 'C': " + employeeMap.floorKey("C"));
// Head and tail views
System.out.println("Head map (before 'C'): " + employeeMap.headMap("C"));
System.out.println("Tail map (after 'C'): " + employeeMap.tailMap("C"));
}
}
class Employee {
private String name;
private int salary;
public Employee(String name, int salary) {
this.name = name;
this.salary = salary;
}
@Override
public String toString() {
return name + "($" + salary + ")";
}
}
4. Concurrent Utility Classes
CountDownLatch
import java.util.concurrent.*;
public class CountDownLatchExample {
public static void main(String[] args) throws InterruptedException {
int workerCount = 3;
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(workerCount);
for (int i = 0; i < workerCount; i++) {
new Thread(new Worker(startSignal, doneSignal, "Worker-" + i)).start();
}
System.out.println("Main thread preparing resources...");
Thread.sleep(2000);
System.out.println("Starting all workers...");
startSignal.countDown(); // Release all workers
System.out.println("Main thread waiting for workers to complete...");
doneSignal.await(); // Wait for all workers to complete
System.out.println("All workers completed. Main thread continuing.");
}
}
class Worker implements Runnable {
private final CountDownLatch startSignal;
private final CountDownLatch doneSignal;
private final String name;
public Worker(CountDownLatch startSignal, CountDownLatch doneSignal, String name) {
this.startSignal = startSignal;
this.doneSignal = doneSignal;
this.name = name;
}
@Override
public void run() {
try {
System.out.println(name + " waiting to start...");
startSignal.await(); // Wait for start signal
System.out.println(name + " working...");
Thread.sleep(1000 + (int)(Math.random() * 2000));
System.out.println(name + " completed work.");
doneSignal.countDown(); // Signal completion
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
CyclicBarrier
import java.util.concurrent.*;
public class CyclicBarrierExample {
public static void main(String[] args) {
int threadCount = 3;
CyclicBarrier barrier = new CyclicBarrier(threadCount,
() -> System.out.println("All threads reached barrier! Continuing..."));
for (int i = 0; i < threadCount; i++) {
new Thread(new Task(barrier, "Task-" + i)).start();
}
}
}
class Task implements Runnable {
private final CyclicBarrier barrier;
private final String name;
public Task(CyclicBarrier barrier, String name) {
this.barrier = barrier;
this.name = name;
}
@Override
public void run() {
try {
System.out.println(name + " starting phase 1");
Thread.sleep(1000 + (int)(Math.random() * 2000));
System.out.println(name + " finished phase 1, waiting at barrier");
barrier.await();
System.out.println(name + " starting phase 2");
Thread.sleep(1000 + (int)(Math.random() * 2000));
System.out.println(name + " finished phase 2, waiting at barrier");
barrier.await();
System.out.println(name + " completed all phases");
} catch (InterruptedException | BrokenBarrierException e) {
Thread.currentThread().interrupt();
}
}
}
Exchanger
import java.util.concurrent.*;
public class ExchangerExample {
public static void main(String[] args) {
Exchanger<String> exchanger = new Exchanger<>();
new Thread(new Producer(exchanger)).start();
new Thread(new Consumer(exchanger)).start();
}
}
class Producer implements Runnable {
private final Exchanger<String> exchanger;
private static int counter = 0;
public Producer(Exchanger<String> exchanger) {
this.exchanger = exchanger;
}
@Override
public void run() {
try {
for (int i = 0; i < 3; i++) {
String data = "Data-" + (++counter);
System.out.println("Producer sending: " + data);
// Exchange data with consumer and get response
String response = exchanger.exchange(data);
System.out.println("Producer received: " + response);
Thread.sleep(1000);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
class Consumer implements Runnable {
private final Exchanger<String> exchanger;
public Consumer(Exchanger<String> exchanger) {
this.exchanger = exchanger;
}
@Override
public void run() {
try {
for (int i = 0; i < 3; i++) {
// Receive data from producer
String data = exchanger.exchange(null);
System.out.println("Consumer received: " + data);
// Process and send response
String response = "Ack-" + data;
System.out.println("Consumer sending: " + response);
// Exchange response with producer
exchanger.exchange(response);
Thread.sleep(1500);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
5. Performance Comparison
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
public class PerformanceComparison {
private static final int THREAD_COUNT = 10;
private static final int OPERATIONS_PER_THREAD = 10000;
public static void main(String[] args) throws InterruptedException {
System.out.println("Performance Comparison:");
// Test HashMap vs ConcurrentHashMap
testMapPerformance();
// Test ArrayList vs CopyOnWriteArrayList
testListPerformance();
// Test Queue implementations
testQueuePerformance();
}
public static void testMapPerformance() throws InterruptedException {
// HashMap with synchronization
Map<Integer, Integer> syncMap = Collections.synchronizedMap(new HashMap<>());
// ConcurrentHashMap
ConcurrentHashMap<Integer, Integer> concurrentMap = new ConcurrentHashMap<>();
long syncTime = testMap(syncMap, "Synchronized HashMap");
long concurrentTime = testMap(concurrentMap, "ConcurrentHashMap");
System.out.printf("ConcurrentHashMap is %.2fx faster%n%n",
(double) syncTime / concurrentTime);
}
public static long testMap(Map<Integer, Integer> map, String name) throws InterruptedException {
long startTime = System.currentTimeMillis();
CountDownLatch latch = new CountDownLatch(THREAD_COUNT);
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadId = i;
new Thread(() -> {
for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {
int key = threadId * OPERATIONS_PER_THREAD + j;
map.put(key, key);
map.get(key);
}
latch.countDown();
}).start();
}
latch.await();
long endTime = System.currentTimeMillis();
System.out.printf("%s: %d ms%n", name, (endTime - startTime));
return endTime - startTime;
}
public static void testListPerformance() throws InterruptedException {
// ArrayList with synchronization
List<Integer> syncList = Collections.synchronizedList(new ArrayList<>());
// CopyOnWriteArrayList
CopyOnWriteArrayList<Integer> copyOnWriteList = new CopyOnWriteArrayList<>();
long syncTime = testList(syncList, "Synchronized ArrayList");
long copyOnWriteTime = testList(copyOnWriteList, "CopyOnWriteArrayList");
System.out.printf("For read-heavy: CopyOnWriteArrayList is better%n");
System.out.printf("For write-heavy: Synchronized ArrayList is %.2fx faster%n%n",
(double) copyOnWriteTime / syncTime);
}
public static long testList(List<Integer> list, String name) throws InterruptedException {
long startTime = System.currentTimeMillis();
CountDownLatch latch = new CountDownLatch(THREAD_COUNT);
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadId = i;
new Thread(() -> {
// 80% reads, 20% writes
for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {
if (j % 5 == 0) {
// Write operation
list.add(threadId * OPERATIONS_PER_THREAD + j);
} else {
// Read operation
if (!list.isEmpty()) {
list.get(list.size() - 1);
}
}
}
latch.countDown();
}).start();
}
latch.await();
long endTime = System.currentTimeMillis();
System.out.printf("%s: %d ms%n", name, (endTime - startTime));
return endTime - startTime;
}
public static void testQueuePerformance() throws InterruptedException {
BlockingQueue<Integer> arrayQueue = new ArrayBlockingQueue<>(1000);
BlockingQueue<Integer> linkedQueue = new LinkedBlockingQueue<>(1000);
ConcurrentLinkedQueue<Integer> concurrentQueue = new ConcurrentLinkedQueue<>();
long arrayTime = testQueue(arrayQueue, "ArrayBlockingQueue");
long linkedTime = testQueue(linkedQueue, "LinkedBlockingQueue");
long concurrentTime = testQueue(concurrentQueue, "ConcurrentLinkedQueue");
System.out.printf("ConcurrentLinkedQueue is %.2fx faster than ArrayBlockingQueue%n%n",
(double) arrayTime / concurrentTime);
}
public static long testQueue(Queue<Integer> queue, String name) throws InterruptedException {
long startTime = System.currentTimeMillis();
CountDownLatch latch = new CountDownLatch(THREAD_COUNT * 2);
// Producers
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadId = i;
new Thread(() -> {
for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {
if (queue instanceof BlockingQueue) {
try {
((BlockingQueue<Integer>) queue).put(threadId * OPERATIONS_PER_THREAD + j);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
} else {
queue.offer(threadId * OPERATIONS_PER_THREAD + j);
}
}
latch.countDown();
}).start();
}
// Consumers
for (int i = 0; i < THREAD_COUNT; i++) {
new Thread(() -> {
for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {
if (queue instanceof BlockingQueue) {
try {
((BlockingQueue<Integer>) queue).take();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
} else {
queue.poll();
}
}
latch.countDown();
}).start();
}
latch.await();
long endTime = System.currentTimeMillis();
System.out.printf("%s: %d ms%n", name, (endTime - startTime));
return endTime - startTime;
}
}
Best Practices
- Choose the right concurrent collection based on your use case
- Use ConcurrentHashMap for high-concurrency maps
- Use CopyOnWriteArrayList for read-heavy, write-rarely scenarios
- Prefer ConcurrentLinkedQueue for non-blocking operations
- Use BlockingQueue for producer-consumer scenarios
- Consider ConcurrentSkipListMap when you need sorted order
- Always understand the consistency guarantees of each collection
Concurrent collections provide efficient thread-safe alternatives to synchronized wrappers, offering better performance and scalability in multi-threaded applications.