【Java】并发容器深度解析

Java并发容器深度解析

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

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

// JDK 7 核心结构finalSegment<K,V>[]segments;// 16个段默认staticfinalclassSegment<K,V>extendsReentrantLock{transientvolatileHashEntry<K,V>[]table;// 每个段的哈希表transientintcount;// 段内元素个数}

核心特点：

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

put操作伪代码：

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

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

// JDK 8+ 核心结构transientvolatileNode<K,V>[]table;// Node数组transientvolatileintsizeCtl;// 控制标志位staticclassNode<K,V>implementsMap.Entry<K,V>{finalinthash;finalKkey;volatileVval;volatileNode<K,V>next;}

革命性改进：

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

put操作源码分析：

finalVputVal(Kkey,Vvalue,booleanonlyIfAbsent){if(key==null||value==null)thrownewNullPointerException();inthash=spread(key.hashCode());intbinCount=0;for(Node<K,V>[]tab=table;;){Node<K,V>f;intn,i,fh;// 1. 初始化：CAS保证单线程创建tableif(tab==null||(n=tab.length)==0)tab=initTable();// 2. 定位桶：CAS获取头节点elseif((f=tabAt(tab,i=(n-1)&hash))==null){if(casTabAt(tab,i,null,newNode<K,V>(hash,key,value,null)))break;// 无竞争直接插入成功}// 3. 协助扩容elseif((fh=f.hash)==MOVED)tab=helpTransfer(tab,f);// 4. 同步块：只锁单个Nodeelse{VoldVal=null;synchronized(f){// 细化到桶级别if(tabAt(tab,i)==f){if(fh>=0){// 链表binCount=1;for(Node<K,V>e=f;;++binCount){Kek;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=newNode<K,V>(hash,key,value,null);break;}}}elseif(finstanceofTreeBin){// 红黑树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)returnoldVal;break;}}}addCount(1L,binCount);returnnull;}

1.3 关键差异对比

二、CopyOnWriteArrayList：写时复制容器

2.1 核心原理

publicclassCopyOnWriteArrayList<E>implementsList<E>{privatetransientvolatileObject[]array;// volatile保证可见性finalObject[]getArray(){returnarray;}finalvoidsetArray(Object[]a){array=a;}}

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

publicbooleanadd(Ee){finalReentrantLocklock=this.lock;lock.lock();// 写操作加锁try{Object[]elements=getArray();intlen=elements.length;Object[]newElements=Arrays.copyOf(elements,len+1);// 复制新数组newElements[len]=e;setArray(newElements);// volatile写，立即可见returntrue;}finally{lock.unlock();}}publicEget(intindex){returnget(getArray(),index);// 读操作无锁，直接访问volatile数组}

2.2 性能特征分析

优势：

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

代价：

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

2.3 适用场景与陷阱

正确场景：

// 事件监听器列表（读远多于写）privatefinalCopyOnWriteArrayList<EventListener>listeners=newCopyOnWriteArrayList<>();publicvoidaddListener(EventListenerlistener){listeners.add(listener);// 写少}publicvoidfireEvent(Eventevent){for(EventListenerlistener:listeners){// 读多，无锁listener.onEvent(event);}}

错误场景：

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

三、BlockingQueue：阻塞队列家族

3.1 接口定义与核心方法

publicinterfaceBlockingQueue<E>extendsQueue<E>{// 阻塞方法voidput(Ee)throwsInterruptedException;// 队列满时阻塞Etake()throwsInterruptedException;// 队列空时阻塞// 超时方法booleanoffer(Ee,longtimeout,TimeUnitunit);Epoll(longtimeout,TimeUnitunit);// 非阻塞方法booleanoffer(Ee);// 失败立即返回falseEpoll();// 失败立即返回null}

3.2 主要实现类对比

3.3 核心实现剖析

LinkedBlockingQueue：双锁分离设计

publicclassLinkedBlockingQueue<E>{privatefinalAtomicIntegercount=newAtomicInteger();// 原子计数privatefinalReentrantLocktakeLock=newReentrantLock();// 消费锁privatefinalConditionnotEmpty=takeLock.newCondition();privatefinalReentrantLockputLock=newReentrantLock();// 生产锁privatefinalConditionnotFull=putLock.newCondition();publicvoidput(Ee)throwsInterruptedException{intc=-1;Node<E>node=newNode<E>(e);finalReentrantLockputLock=this.putLock;finalAtomicIntegercount=this.count;putLock.lockInterruptibly();// 只获取put锁try{while(count.get()==capacity){notFull.await();// 队列满时等待}enqueue(node);// 入队c=count.getAndIncrement();// 原子计数+1if(c+1<capacity)notFull.signal();// 唤醒生产者}finally{putLock.unlock();}// 关键：c==0表示队列由空变非空，唤醒消费者if(c==0)signalNotEmpty();}}

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

SynchronousQueue：零容量队列

// 线程配对传输，无存储空间SynchronousQueue<Integer>queue=newSynchronousQueue<>();// 线程A：put会阻塞直到有线程takenewThread(()->{try{queue.put(1);// 等待消费者}catch(InterruptedExceptione){}}).start();// 线程B：take会阻塞直到有线程putnewThread(()->{try{Integervalue=queue.take();// 立即获得1}catch(InterruptedExceptione){}}).start();

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

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

publicclassDataProcessor{privatefinalBlockingQueue<Task>queue=newLinkedBlockingQueue<>(100);// 有界缓冲// 生产者publicvoidproduce(Tasktask){booleansubmitted=queue.offer(task);// 非阻塞if(!submitted){// 队列满，降级处理log.warn("Queue full, dropping task: {}",task);// 或执行背压策略// handleBackPressure(task);}}// 消费者（多线程）publicvoidconsume(){while(running){try{Tasktask=queue.poll(1,TimeUnit.SECONDS);if(task!=null){process(task);// 处理任务}}catch(InterruptedExceptione){Thread.currentThread().interrupt();break;}}}}// 优雅关闭publicvoidshutdown()throwsInterruptedException{running=false;// 等待队列消费完毕while(!queue.isEmpty()){Thread.sleep(100);}executor.shutdown();executor.awaitTermination(60,TimeUnit.SECONDS);}

四、选型指南与最佳实践

4.1 并发Map选型

// 高并发读写ConcurrentHashMap<String,Object>map=newConcurrentHashMap<>();// 写极少，读极多（配置类）Map<String,String>config=newConcurrentHashMap<>();// 或不可变MapMap<String,String>immutableConfig=Map.copyOf(initialMap);// 统计计数ConcurrentHashMap<String,LongAdder>counter=newConcurrentHashMap<>();counter.computeIfAbsent(key,k->newLongAdder()).increment();

4.2 并发List选型

// 读多写少（事件监听）CopyOnWriteArrayList<Listener>listeners=newCopyOnWriteArrayList<>();// 读写均衡：Collections.synchronizedList + 手动同步List<String>syncList=Collections.synchronizedList(newArrayList<>());// 高性能无锁读：immutableList + volatilevolatileList<String>cachedList=Collections.emptyList();publicvoidupdateList(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(Stringkey:map.keySet()){map.remove(key);// ConcurrentModificationException}// ✅ 正确：使用迭代器或并发方法map.keySet().removeIf(k->shouldRemove(k));

5.2 CopyOnWriteArrayList陷阱

// ❌ 错误：高频修改for(inti=0;i<1000;i++){list.add(i);// 内存爆炸！}// ✅ 正确：批量修改List<Integer>temp=newArrayList<>(list);temp.addAll(batchData);list=newCopyOnWriteArrayList<>(temp);// 一次性替换// ❌ 错误：依赖实时一致性if(list.size()>0){// size可能已变，但迭代是安全的list.forEach(System.out::println);}

总结

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