当前位置：首页 > backend >正文

Java并发编程实战 Day 7：并发集合类详解

backend 2025/6/29 15:08:55

【Java并发编程实战 Day 7】并发集合类详解

开篇

欢迎来到"Java并发编程实战"系列的第7天！今天我们将重点探讨并发集合类（如ConcurrentHashMap和CopyOnWriteArrayList），它们是Java并发编程中的核心组件之一。通过本篇文章，你将掌握这些并发集合的使用方法、实现原理以及最佳实践，为处理高并发场景提供解决方案。

注意：本文适用于有一定多线程基础的开发者，建议先复习前6天的内容以更好地理解今天的主题。

理论基础

什么是并发集合类？

并发集合类是指那些专门为多线程环境设计的集合类型，它们在保证线程安全的同时，尽可能地提升了性能。常见的并发集合类包括：

ConcurrentHashMap
CopyOnWriteArrayList
BlockingQueue及其子类（将在后续章节讲解）

相比传统的同步集合（如Collections.synchronizedMap()包装的集合），并发集合采用更细粒度的锁机制或无锁算法，从而减少线程间的竞争，提高并发性能。

并发集合的核心实现原理

分段锁机制：例如ConcurrentHashMap早期版本通过分段锁（Segment Lock）来实现部分区域的并发写操作，减少了锁的粒度。
CAS无锁算法：从Java 8开始，ConcurrentHashMap改用基于CAS（Compare-And-Swap）的无锁算法，进一步优化了性能。
写时复制：CopyOnWriteArrayList在每次修改时会创建底层数组的新副本，读操作完全无锁，适用于读多写少的场景。

适用场景

ConcurrentHashMap

高频读写场景：如缓存系统中需要频繁对数据进行增删改查。
统计计数器：多个线程同时更新某些计数值。

CopyOnWriteArrayList

读多写少场景：如用户权限列表、配置文件管理等。
事件监听器注册表：监听器通常只在初始化阶段添加或移除，运行时主要为读取。

代码实践

示例1：使用ConcurrentHashMap实现线程安全的缓存

import java.util.concurrent.ConcurrentHashMap;public class CacheExample {public static void main(String[] args) {// 创建一个线程安全的ConcurrentHashMapConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>();// 模拟多个线程同时写入数据Thread writer1 = new Thread(() -> {for (int i = 0; i < 5; i++) {cache.put("Key" + i, "Value" + i);System.out.println("Writer1 added Key" + i);}});Thread writer2 = new Thread(() -> {for (int i = 5; i < 10; i++) {cache.put("Key" + i, "Value" + i);System.out.println("Writer2 added Key" + i);}});// 启动线程writer1.start();writer2.start();try {writer1.join();writer2.join();} catch (InterruptedException e) {e.printStackTrace();}// 打印最终结果System.out.println("Final Cache Content: " + cache);}
}

示例2：使用CopyOnWriteArrayList实现线程安全的事件监听器

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;public class EventListenerExample {public static void main(String[] args) {List<String> listeners = new CopyOnWriteArrayList<>();// 模拟注册监听器Thread registerThread = new Thread(() -> {for (int i = 0; i < 5; i++) {listeners.add("Listener" + i);System.out.println("Registered Listener" + i);}});// 模拟触发事件Thread triggerThread = new Thread(() -> {for (String listener : listeners) {System.out.println("Triggered " + listener);}});registerThread.start();triggerThread.start();try {registerThread.join();triggerThread.join();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("All Listeners: " + listeners);}
}

实现原理

ConcurrentHashMap源码分析

内部结构：ConcurrentHashMap由多个哈希桶组成，每个桶独立加锁。
put()方法流程：
- 计算哈希值并定位到对应的桶。
- 如果桶为空，则尝试通过CAS插入新节点。
- 如果桶非空且未加锁，则获取锁后插入。
扩容机制：当负载因子超过阈值时，会动态扩容，扩容过程中其他线程可以协助迁移数据。

CopyOnWriteArrayList源码分析

读操作：直接返回底层数组的快照，无需加锁。
写操作：每次修改都会创建数组的新副本，确保读操作不受影响。

性能测试

我们对比了ConcurrentHashMap与Collections.synchronizedMap()在高并发写入场景下的性能表现，测试代码如下：

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;public class PerformanceTest {private static final int THREAD_COUNT = 10;private static final int OPERATIONS_PER_THREAD = 10000;public static void main(String[] args) throws InterruptedException {Map<String, String> syncMap = Collections.synchronizedMap(new HashMap<>());Map<String, String> concurrentMap = new ConcurrentHashMap<>();ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);long startTime, endTime;// 测试synchronizedMapstartTime = System.nanoTime();for (int i = 0; i < THREAD_COUNT; i++) {executor.submit(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {syncMap.put("Key" + j, "Value" + j);}});}executor.shutdown();executor.awaitTermination(1, TimeUnit.MINUTES);endTime = System.nanoTime();System.out.println("SynchronizedMap Time: " + (endTime - startTime) / 1e6 + " ms");executor = Executors.newFixedThreadPool(THREAD_COUNT);// 测试ConcurrentHashMapstartTime = System.nanoTime();for (int i = 0; i < THREAD_COUNT; i++) {executor.submit(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {concurrentMap.put("Key" + j, "Value" + j);}});}executor.shutdown();executor.awaitTermination(1, TimeUnit.MINUTES);endTime = System.nanoTime();System.out.println("ConcurrentHashMap Time: " + (endTime - startTime) / 1e6 + " ms");}
}