Java并发容器深度解析

一、ConcurrentHashMap：并发哈希表的演进

1.1 JDK 7：分段锁（Segment）机制

// JDK 7 核心结构
final Segment<K,V>[] segments; // 16个段默认

static final class Segment<K,V> extends ReentrantLock {
    transient volatile HashEntry<K,V>[] table; // 每个段的哈希表
    transient int count; // 段内元素个数
}

核心特点：

• 锁分段：将哈希表分为16个Segment，每个Segment独立加锁
• 写操作：只锁定当前Segment，其他Segment可并发访问
• 读操作：无锁（volatile保证可见性）
• 缺点：Segment数组大小固定，最多支持16个并发写

put操作伪代码：

public V put(K key, V value) {
    int hash = hash(key);
    Segment<K,V> s = segmentForHash(hash); // 定位到Segment
    s.lock(); // 只锁当前段
    try {
        // 在段内执行put...
    } finally {
        s.unlock();
    }
}

---

1.2 JDK 8+：CAS + synchronized + 红黑树

// JDK 8+ 核心结构
transient volatile Node<K,V>[] table; // Node数组
transient volatile int sizeCtl; // 控制标志位

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
}

革命性改进：

• 锁粒度细化：每个Node头节点作为锁（synchronized）
• CAS无锁化：初始化、扩容等关键步骤使用CAS
• 红黑树优化：链表长度>8转红黑树，提升查询性能至O(logn)

put操作源码分析：

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());
    int binCount = 0;
    
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        
        // 1. 初始化：CAS保证单线程创建table
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        
        // 2. 定位桶：CAS获取头节点
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
                break; // 无竞争直接插入成功
        }
        
        // 3. 协助扩容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        
        // 4. 同步块：只锁单个Node
        else {
            V oldVal = null;
            synchronized (f) { // 细化到桶级别
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) { // 链表
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            if (e.hash == hash && ((ek = e.key) == key || ...)) {
                                oldVal = e.val;
                                if (!onlyIfAbsent) e.val = value;
                                break;
                            }
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key, value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) { // 红黑树
                        Node<K,V> p;
                        binCount = 2;
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent) p.val = value;
                        }
                    }
                }
            }
            
            // 5. 链表转红黑树
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null) return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}

---

1.3 关键差异对比

特性	JDK 7 Segment	JDK 8+ Node + CAS
锁粒度	16个Segment	每个Node头节点
并发度	最多16	理论无限制（数组大小）
查询复杂度	O(n) 链表	O(logn) 红黑树
扩容机制	Segment独立扩容	多线程协助扩容
内存占用	较高（Segment对象）	更轻量
适用场景	低并发写入	高并发读写

---

二、CopyOnWriteArrayList：写时复制容器

2.1 核心原理

public class CopyOnWriteArrayList<E> implements List<E> {
    private transient volatile Object[] array; // volatile保证可见性
    
    final Object[] getArray() { return array; }
    final void setArray(Object[] a) { array = a; }
}

设计哲学：读操作无锁，写操作复制新数组

public boolean add(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock(); // 写操作加锁
    try {
        Object[] elements = getArray();
        int len = elements.length;
        Object[] newElements = Arrays.copyOf(elements, len + 1); // 复制新数组
        newElements[len] = e;
        setArray(newElements); // volatile写，立即可见
        return true;
    } finally {
        lock.unlock();
    }
}

public E get(int index) {
    return get(getArray(), index); // 读操作无锁，直接访问volatile数组
}

---

2.2 性能特征分析

优势：

• 读性能极高：无锁，接近普通ArrayList
• 迭代安全：遍历的是快照，不受写操作影响
• 内存一致性：volatile保证立即可见

代价：

• 写性能差：每次复制整个数组，O(n)复杂度
• 内存占用高：同时存在两个数组副本
• 数据延迟：读操作可能读取到旧数据（弱一致性）

---

2.3 适用场景与陷阱

正确场景：

// 事件监听器列表（读远多于写）
private final CopyOnWriteArrayList<EventListener> listeners = 
    new CopyOnWriteArrayList<>();

public void addListener(EventListener listener) {
    listeners.add(listener); // 写少
}

public void fireEvent(Event event) {
    for (EventListener listener : listeners) { // 读多，无锁
        listener.onEvent(event);
    }
}

错误场景：

// 高频写入场景！绝对避免！
CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();
for (int i = 0; i < 100000; i++) {
    list.add(i); // 每次复制，内存和CPU爆炸
}

---

三、BlockingQueue：阻塞队列家族

3.1 接口定义与核心方法

public interface BlockingQueue<E> extends Queue<E> {
    // 阻塞方法
    void put(E e) throws InterruptedException; // 队列满时阻塞
    E take() throws InterruptedException;      // 队列空时阻塞
    
    // 超时方法
    boolean offer(E e, long timeout, TimeUnit unit);
    E poll(long timeout, TimeUnit unit);
    
    // 非阻塞方法
    boolean offer(E e); // 失败立即返回false
    E poll();           // 失败立即返回null
}

---

3.2 主要实现类对比

实现类	底层结构	容量	锁机制	适用场景
ArrayBlockingQueue	数组	有界	全局ReentrantLock	固定大小缓冲区
LinkedBlockingQueue	链表	可选有界	双锁（put/take分离）	高并发吞吐量
PriorityBlockingQueue	堆	无界	全局ReentrantLock	优先级任务调度
SynchronousQueue	无存储	0容量	CAS/TransferQueue	直接传递，线程配对
DelayQueue	PriorityQueue	无界	全局ReentrantLock	延迟任务调度
LinkedTransferQueue	链表	无界	CAS + 自旋	高性能传输

---

3.3 核心实现剖析

LinkedBlockingQueue：双锁分离设计

public class LinkedBlockingQueue<E> {
    private final AtomicInteger count = new AtomicInteger(); // 原子计数
    
    private final ReentrantLock takeLock = new ReentrantLock(); // 消费锁
    private final Condition notEmpty = takeLock.newCondition();
    
    private final ReentrantLock putLock = new ReentrantLock(); // 生产锁
    private final Condition notFull = putLock.newCondition();
    
    public void put(E e) throws InterruptedException {
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        
        putLock.lockInterruptibly(); // 只获取put锁
        try {
            while (count.get() == capacity) {
                notFull.await(); // 队列满时等待
            }
            enqueue(node); // 入队
            c = count.getAndIncrement(); // 原子计数+1
            if (c + 1 < capacity) notFull.signal(); // 唤醒生产者
        } finally {
            putLock.unlock();
        }
        
        // 关键：c==0表示队列由空变非空，唤醒消费者
        if (c == 0) signalNotEmpty();
    }
}

性能优势：put和take操作使用不同锁，吞吐量比ArrayBlockingQueue高2-3倍。

---

SynchronousQueue：零容量队列

// 线程配对传输，无存储空间
SynchronousQueue<Integer> queue = new SynchronousQueue<>();

// 线程A：put会阻塞直到有线程take
new Thread(() -> {
    try {
        queue.put(1); // 等待消费者
    } catch (InterruptedException e) {}
}).start();

// 线程B：take会阻塞直到有线程put
new Thread(() -> {
    try {
        Integer value = queue.take(); // 立即获得1
    } catch (InterruptedException e) {}
}).start();

底层实现：使用TransferStack/TransferQueue算法，通过CAS实现线程配对，性能极高。

---

3.4 实战：生产者-消费者模式

public class DataProcessor {
    private final BlockingQueue<Task> queue = 
        new LinkedBlockingQueue<>(100); // 有界缓冲
    
    // 生产者
    public void produce(Task task) {
        boolean submitted = queue.offer(task); // 非阻塞
        if (!submitted) {
            // 队列满，降级处理
            log.warn("Queue full, dropping task: {}", task);
            // 或执行背压策略
            // handleBackPressure(task);
        }
    }
    
    // 消费者（多线程）
    public void consume() {
        while (running) {
            try {
                Task task = queue.poll(1, TimeUnit.SECONDS);
                if (task != null) {
                    process(task); // 处理任务
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

// 优雅关闭
public void shutdown() throws InterruptedException {
    running = false;
    // 等待队列消费完毕
    while (!queue.isEmpty()) {
        Thread.sleep(100);
    }
    executor.shutdown();
    executor.awaitTermination(60, TimeUnit.SECONDS);
}

---

四、选型指南与最佳实践

4.1 并发Map选型

// 高并发读写
ConcurrentHashMap<String, Object> map = new ConcurrentHashMap<>();

// 写极少，读极多（配置类）
Map<String, String> config = new ConcurrentHashMap<>();
// 或不可变Map
Map<String, String> immutableConfig = Map.copyOf(initialMap);

// 统计计数
ConcurrentHashMap<String, LongAdder> counter = new ConcurrentHashMap<>();
counter.computeIfAbsent(key, k -> new LongAdder()).increment();

4.2 并发List选型

// 读多写少（事件监听）
CopyOnWriteArrayList<Listener> listeners = new CopyOnWriteArrayList<>();

// 读写均衡：Collections.synchronizedList + 手动同步
List<String> syncList = Collections.synchronizedList(new ArrayList<>());

// 高性能无锁读：immutableList + volatile
volatile List<String> cachedList = Collections.emptyList();

public void updateList(List<String> newList) {
    this.cachedList = List.copyOf(newList); // 原子更新
}

4.3 队列选型决策树

需要阻塞？
├─ 是 → 需要延迟/优先级？
│   ├─ 延迟 → DelayQueue
│   ├─ 优先级 → PriorityBlockingQueue
│   └─ 普通 → 需要直接传递？
│       ├─ 是 → SynchronousQueue
│       └─ 否 → 有界？ArrayBlockingQueue : LinkedBlockingQueue
└─ 否 → 需要并发安全？
    ├─ 是 → ConcurrentLinkedQueue
    └─ 否 → ArrayDeque/LinkedList

---

五、性能陷阱与规避

5.1 ConcurrentHashMap陷阱

// ❌ 错误：复合操作非原子
if (map.get(key) == null) {
    map.put(key, computeValue()); // 可能覆盖其他线程的值
}

// ✅ 正确：使用原子方法
map.computeIfAbsent(key, k -> computeValue());

// ❌ 错误：遍历同时修改会抛异常
for (String key : map.keySet()) {
    map.remove(key); // ConcurrentModificationException
}

// ✅ 正确：使用迭代器或并发方法
map.keySet().removeIf(k -> shouldRemove(k));

5.2 CopyOnWriteArrayList陷阱

// ❌ 错误：高频修改
for (int i = 0; i < 1000; i++) {
    list.add(i); // 内存爆炸！
}

// ✅ 正确：批量修改
List<Integer> temp = new ArrayList<>(list);
temp.addAll(batchData);
list = new CopyOnWriteArrayList<>(temp); // 一次性替换

// ❌ 错误：依赖实时一致性
if (list.size() > 0) {
    // size可能已变，但迭代是安全的
    list.forEach(System.out::println);
}

---

总结

容器	核心机制	适用场景	禁用场景
ConcurrentHashMap	CAS + synchronized	高并发读写	强一致性复合操作
CopyOnWriteArrayList	写时复制	读多写少（监听列表）	高频修改、大数据量
BlockingQueue	锁/Condition	生产者-消费者	无界队列+高速生产

金句：并发容器没有银弹，理解其内部机制是正确选型的前提。在性能敏感场景，务必通过JMH压测验证选择。