并发安全容器
·
安全集合
Vector集合
元老级别的集合，至于它为为什么是线程安全的集合，很简单每个方法都加了synchronized，所以对它的操作都是安全的，但也就是这个问题每个方法都加了锁，所以同步开销很大，项目中现在已经基本看不到了。
这里也贴一下源码更为直观：
// 1. 添加元素 - 线程安全
public synchronized boolean add(E e) {
modCount++;
ensureCapacityHelper(elementCount + 1); // 容量检查
elementData[elementCount++] = e; // 添加元素
return true;
}
// 2. 获取元素 - 线程安全
public synchronized E get(int index) {
if (index >= elementCount)
throw new ArrayIndexOutOfBoundsException(index);
return elementData(index); // 返回指定位置的元素
}
// 3. 删除元素 - 线程安全
public synchronized E remove(int index) {
modCount++;
if (index >= elementCount)
throw new ArrayIndexOutOfBoundsException(index);
E oldValue = elementData(index);
int numMoved = elementCount - index - 1;
if (numMoved > 0)
System.arraycopy(elementData, index+1, elementData, index, numMoved);
elementData[--elementCount] = null; // 清除引用，帮助GC
return oldValue;
}
SynchronizedList包装类
Vector底层是数组实现，如果想使用链表实现的数据结构集合，这个时候就可以使用SynchronizedList。
SynchronizedList 是 Java 集合框架中的一个包装类，继承了List接口，但是它本身并不实现 List 接口中的方法，而是将另一个普通的 List 对象（比如 ArrayList 或 LinkedList）包装起来，为其提供线程安全的访问能力。
因为SynchronizedList 是Collections中的一个静态内部类，所以不能直接通过 new SynchronizedList() 来创建它。必须通过 java.util.Collections 工具类来创建。
List<Object> syncArrayList = Collections.synchronizedList(new ArrayList<>());
List<Object> syncLinkedList = Collections.synchronizedList(new LinkedList<>());
被包装起来的集合，所有的方法调用都被装饰了，调用也就是线程安全的。
public class SynchronizedListCreation {
public static void main(String[] args) {
// 方式1：包装 ArrayList
List<String> syncList1 = Collections.synchronizedList(new ArrayList<>());
// 方式2：包装 LinkedList
List<Integer> syncList2 = Collections.synchronizedList(new LinkedList<>());
// 方式3：包装已有列表
List<Object> existingList = new ArrayList<>();
existingList.add("item1");
List<Object> syncList3 = Collections.synchronizedList(existingList);
// 基本操作 - 线程安全
syncList1.add("hello");
syncList1.add("world");
String first = syncList1.get(0); // 线程安全的读取
System.out.println("Size: " + syncList1.size());
System.out.println("First element: " + first);
}
}
看下源码它是如何进行装饰而到达线程安全
// Collections.synchronizedList 方法源码思路
public static <T> List<T> synchronizedList(List<T> list) {
// 这儿其实就是区分底层是数组实现还是链表实现，数组实现的ArrayList支持随机访问的所以实现RandomAccess返回的就是SynchronizedRandomAccessList
// LinkedList链表实现不能随机访问，所以返回的是SynchronizedList
return (list instanceof RandomAccess ?
new SynchronizedRandomAccessList<>(list) :
new SynchronizedList<>(list));
}
// SynchronizedList 核心实现,属于静态内部类
static class SynchronizedList<E>
extends SynchronizedCollection<E>
implements List<E> {
final List<E> list; // 被装饰的原始列表
SynchronizedList(List<E> list) {
super(list);
this.list = list;
}
// 所有方法都使用同一个锁对象（mutex）
public E get(int index) {
synchronized (mutex) { return list.get(index); }
}
public E set(int index, E element) {
synchronized (mutex) { return list.set(index, element); }
}
public void add(int index, E element) {
synchronized (mutex) { list.add(index, element); }
}
public E remove(int index) {
synchronized (mutex) { return list.remove(index); }
}
public int indexOf(Object o) {
synchronized (mutex) { return list.indexOf(o); }
}
public int lastIndexOf(Object o) {
synchronized (mutex) { return list.lastIndexOf(o); }
}
}
可以看到，每个方法进行包装后都加了一个synchronized 锁，接着就调用原来集合的具体实现方法完成数据操作。所以包装过后的集合调用的方法都是线程安全的。
那么这个锁的对象mutex是什么玩意儿呢？
static class SynchronizedList<E>
extends SynchronizedCollection<E>
implements List<E> {
final List<E> list; // 传进来的集合
SynchronizedList(List<E> list) {
super(list); // 调用父类构造器
this.list = list;
}
}
// SynchronizedList 继承了 SynchronizedCollection ,所以调了SynchronizedCollection 的构造器
static class SynchronizedCollection<E>
implements Collection<E>, Serializable {
final Collection<E> c;
final Object mutex; // 同步对象
SynchronizedCollection(Collection<E> c) {
this.c = Objects.requireNonNull(c);
mutex = this; // 当前对象，也就是 synchronizedCollection 实例本身
}
}
所以说上面的就等价于synchronized (mutex) == synchronized (this)，但是为啥设计者不直接写synchronized (this)呢？，定义mutex主要是可以自定义指定锁的对象，不一定是this对象，方便扩展么。
源码中其实还有一个静态方法可以传入参数mutex指定锁的对象。但是啊这个方法并没有暴露到外部使用，所以我们只能使用第一个方法去创建实例，也就是mutex无特殊情况就是this。
public static <T> List<T> synchronizedList(List<T> list) {
return (list instanceof RandomAccess ?
new SynchronizedRandomAccessList<>(list) :
new SynchronizedList<>(list));
}
static <T> List<T> synchronizedList(List<T> list, Object mutex) {
return (list instanceof RandomAccess ?
new SynchronizedRandomAccessList<>(list, mutex) :
new SynchronizedList<>(list, mutex));
}
注意使用线程安全问题！！！
private static List<String> syncList = Collections.synchronizedList(new ArrayList<>());
// 危险的迭代方式 - 可能抛出 ConcurrentModificationException
public static void unsafeIteration() {
// 虽然单个操作是线程安全的，但复合操作不是
for (String item : syncList) { // 这里没有同步！
System.out.println(item);
// 如果在迭代过程中其他线程修改了列表，会抛出异常
}
}
为什么get()方法虽然加了锁，为啥还是会抛出ConcurrentModificationException，比如：A线程get(0)元素后就释放锁了，执行get(1)之前，B线程就抢到锁然了后修改list数据，A线程就会抛出ConcurrentModificationException，专业术语叫快速失败来着(fast-fail)
正确的安全访问方式：
private static List<String> syncList = Collections.synchronizedList(new ArrayList<>());
// 正确的迭代方式1：手动同步整个迭代过程
public static void safeIteration1() {
synchronized (syncList) { // 关键：手动同步
for (String item : syncList) {
System.out.println(item);
// 可以安全地进行耗时操作
try {
Thread.sleep(10);
} catch (InterruptedException e) {
}
}
}
}
// 正确的迭代方式2：复制后迭代（适用于读多写少）
public static void safeIteration3() {
List<String> copy;
synchronized (syncList) {
copy = new ArrayList<>(syncList); // 创建快照
}
// 在副本上安全迭代，无锁
for (String item : copy) {
System.out.println(item);
}
}
第一种方式是对整个集合加了锁，所以期间不会有第二个线程获取到锁去修改集合。
第二种方式是访问直接新建了一个副本，访问在副本上读取数据，原集合的修改就不会影响副本的数据了。
CopyOnWriteArrayList
这里内容参考了大佬的原文，原文链接为Java并发8–并发安全容器详解 - 技术栈
首先理一下为啥需要开发一个这样的容器？
回顾一下vector和SynchronizedList可以发现，不管是读还是写都是一股脑的加锁，如果存在只有读没有写的并发环境，或者读多写少的环境，岂不是加锁解锁加锁解锁操作太频繁了，而且synchronized加锁会涉及用户态和内核态的切换，很消耗性能的，所以大佬们又又又设计了一个新的并发容器copyOnWriteArrayList来满足这种环境的需求。
copyOnWriteArrayList顾名思义就是写时复制数组集合，写时复制啥意思嘞？
先来了解一下计算机中常用的一种设计思想写时复制：
多个调用者最初可以共享同一份资源（如数据、内存页），只有在其中一个调用者试图修改这份资源时，系统才会真正复制一份副本给该调用者，然后修改在这个副本上进行。在此之前，所有调用者都只是“只读”地访问原始资源。
简单来说，就是 “不写不复制，要写才复制”。
在redis的数据持久化原理、操作系统fork操作、以及MySQL数据库的MVCC中，你都可以看到写时复制这一设计思想。
接下来看下copyOnWriteArrayList是怎么实现的。
数据结构
// jdk21版本
public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
final transient Object lock = new Object(); // 源码中加锁使用的是synchronized对这个lock对象进行加锁
private transient volatile Object[] array;
}
// jdk1.8版本
public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
final transient ReentrantLock lock = new ReentrantLock(); // 使用的是ReentrantLock
private transient volatile Object[] array;
}
我看了下大佬写的文章中源码是8版本的，我自己用的21版本的，所以接下的的源码解释我会用21版本的，对8感兴趣就看下大佬的原文。Java并发8–并发安全容器详解 - 技术栈
至于为什么用ReentrantLock后又改为synchronized（应该是前几个版本就改了），估计是现在的synchronized性能和ReentrantLock到差不差了，就像18之后偏向锁就被废弃了这些改动似的，具体原因的话我也没找到，知道的朋友评论区说一下，我也涨涨脑子。
获取元素
public E get(int index) {
return elementAt(getArray(), index);
}
static <E> E elementAt(Object[] a, int index) {
return (E) a[index]; // 直接返回数下标的元素了
}
源码也可以看到整个获取元素流程都没有去加锁，所以说并发读性能很高。
添加元素
public boolean add(E e) {
synchronized (lock) { // 加锁添加元素
Object[] es = getArray();
int len = es.length;
es = Arrays.copyOf(es, len + 1); // 复制数组
es[len] = e;
setArray(es); // 修改引用指向
return true;
}
}
final void setArray(Object[] a) {
array = a; // 引用指向最新数组
}
添加元素的线程需要先获取锁，然后直接拷贝一个新数组，最后设置值后将引用指向新的数组位置。这里实现和1.8的不太一样，感兴趣看下大佬原文Java并发8–并发安全容器详解 - 技术栈
修改元素
public E set(int index, E element) {
synchronized (lock) { // 加锁
Object[] es = getArray();
E oldValue = elementAt(es, index); // 获取需要修改下标的旧值
if (oldValue != element) { // 旧值和新值如果不相等才会克隆一个新的数组，然后修改值
es = es.clone();
es[index] = element;
}
setArray(es); // 改变引用
return oldValue; // 返回旧值
}
}
修改元素流程：
- 修改线程需要先获取到锁
- 判断修改的值和旧值是否相等
- 不相等才会克隆一个新的数组，修改原值
- 修改原数组引用，改为新数组的引用
这里也是和1.8的逻辑不一样。
删除元素
public E remove(int index) {
synchronized (lock) { // 老规矩加锁
Object[] es = getArray();
int len = es.length;
E oldValue = elementAt(es, index); // 获取旧值
int numMoved = len - index - 1; // 计算删除值的下标
Object[] newElements;
if (numMoved == 0) // 删除元素是最后一个值
newElements = Arrays.copyOf(es, len - 1); // 直接拷贝数组前面的元素到新数组完事了
else {
newElements = new Object[len - 1]; // 不然就先开辟一个新的空间,长度len-1
System.arraycopy(es, 0, newElements, 0, index); // 拷贝原数组中删除元素下标的前半段数据
System.arraycopy(es, index + 1, newElements, index,numMoved);// 拷贝原数组中删除元素下标的后半段数据，最终就得到了删除后的完整数组
}
setArray(newElements); // 修改引用
return oldValue;
}
}
总结
copyOnWriteArrayList于vector和SynchronizedList区别在于，copyOnWriteArrayList的获取元素方法是没有加锁的，所以并发环境下获取元素很快，它只对曾删改加锁了，而且基本都是先拷贝出一个新数组，再新数组上进行操作，最后修改引用指向新数组。就是因为这种读和写同时进行时，写在新数组上执行，读在旧数组上获取，读写分离，所以CopyOnWriteArrayList的迭代器并不是fail-fast的，即代表着当有一条线程在通过迭代器遍历一个CopyOnWriteArrayList对象时，另外一条线程对该容器进行了写操作，不会对使用迭代器遍历容器的线程产生影响，而ArrayList容器，迭代器则是fail-fast的，当一条线程使用迭代器遍历数据，另外一条执行修改操作时，迭代器线程会抛出异常。
关于写时复制的容器，优势比较明显，其内部充分运用了读写分离的思想提升了容器的整体并发吞吐量，以及避免了并发修改抛出异常。
但是也存在两个致命的缺陷：
- 内存占用问题：因为CopyOnWrite容器每次在发生修改时都会复制一个新的数组，所以当数组数据过大时对内存消耗比较高。
- 数据不一致性问题： CopyOnWrite容器保证的是最终一致性，一条线程在执行修改操作，另一条线程在执行读取操作，读取的线程并不能看到最新的数据，就算修改操作执行了setArray()方法将指向改成了新数组，原本读取的线程也不能看到最新的数据。因为读取线程在执行读操作时并不是直接访问成员array完成的，而是通过getArray()方法的形式获取到的数组数据，在getArray()方法执行完成之后，读取数据的线程拿到的引用已经是旧数组的地址了，之后就算修改成员array的指向也不会影响get的访问。
⚠️ CopyOnWrite写时复制容器提升的只是读操作的吞吐量，而整个容器的写操作还是基于同一把独占锁保证的线程安全，所以如果需要频繁执行写操作的场景，并不适合用CopyOnWrite容器，同时还会因为复制带来的内存、时间开销导致性能下降。
CopyOnWriteArraySet
CopyOnWriteArraySet的底层是CopyOnWriteArrayList，无非就是set会增加重复值的判断，在整体逻辑上，还是拷贝出新数组，在新数组上增删改，最后修改原引用这套逻辑。
public class CopyOnWriteArraySet<E> extends AbstractSet<E> implements java.io.Serializable {
// 内部存储数据的结构,使用的就是CopyOnWriteArrayList
private final CopyOnWriteArrayList<E> al;
// 构造器
public CopyOnWriteArraySet() {
al = new CopyOnWriteArrayList<E>();
}
}
添加元素
public boolean add(E e) {
return al.addIfAbsent(e); // 调用addIfAbsent方法
}
public boolean addIfAbsent(E e) { //注意这个方法还没有加锁，也就时多个线程可以同时执行这段逻辑
Object[] snapshot = getArray();
return indexOfRange(e, snapshot, 0, snapshot.length) < 0 // indexOfRange这方法很简单，就是遍历数组看下有没有相同的值已经存在，不存在就返回-1
&& addIfAbsent(e, snapshot); // 前面判断出元素不存在就执行添加元素逻辑
}
private boolean addIfAbsent(E e, Object[] snapshot) { // e需要添加的元素，snapshot看上面就是添加元素前的原数组
synchronized (lock) { // 加锁
Object[] current = getArray(); // 原数组
int len = current.length; // 原数组长度
// 判断当前线程获取到的原数组和实际的原数组相不相等为啥要进行这个判断呢，举个例子
// A,B线程同时需要进行元素的添加，上一个方法不是没加锁么，他们都执行了检查并且元素都不存在，都进入了这个方法，此时A先获取到锁，B就要阻塞等待，等A执行完了已经在原数组上添加好元素后，释放锁
// 此时B被唤醒，获取到锁，但是注意snapshot之前就获取了并且放在自己的栈里面，A线程的添加修改B线程时感知不到的，所以这里需要在判断一下，至于为什么current是最新的，是因为array定义时指定了volatile
if (snapshot != current) {
int common = Math.min(snapshot.length, len);
for (int i = 0; i < common; i++) // 再判断一下元素存不存在，这里考虑了删除操作所以取小长度进行遍历，避免数组越界
if (current[i] != snapshot[i]
&& Objects.equals(e, current[i]))
return false;
if (indexOfRange(e, current, common, len) >= 0) // 这里还要判断，考虑的是添加操作，所以他只判断了添加后的区间有没有重复元素
return false;
}
Object[] newElements = Arrays.copyOf(current, len + 1); //拷贝数组，添加元素
newElements[len] = e;
setArray(newElements); // 修改引用
return true;
}
}
一开始不准备讲CopyOnWriteArraySet源码的，因为它实现和CopyOnWriteArrayList思想一样，并且底层使用的也是CopyOnWriteArrayList，但是看了下源码，我觉得它的设计思想很缜密，值得学习一下。
有没有注意到addIfAbsent(E e)这个方法没有加锁，而是等到判断是否重复添加后才加锁，我是疑惑的，为啥不这里直接加锁，，不然并发下不就出问题了么，但是仔细一想，如果多个线程并发的情况下，同时去检查，对于已经存在的元素值的线程就返回了，只有不存在重复值的线程继续往下执行去争抢锁，这并发量不就上去了么，秒！
但是像我这种菜鸡，后面加锁估计就不会考虑到并发的问题，（当然我估计也想不到前面不加锁，直接一股脑加锁了）直接新建数组添加元素完事了，很显然这种写法肯定会出问题的，所以大佬后面又对上一步没有加锁带来的问题进行了补救，才有了snapshot != current这一步。
多看大佬写的代码不是没有道理的，特别是像我这种菜鸡，开发中运用这些设计经验，对减少系统bug，提升程序性能肯定是有帮助的，感谢大佬们的贡献。
CopyOnWriteArraySet其他方法就不讲了，看了下remove方法设计思想和这个差不多的，就不再赘述了。
ConcurrentLinkedQueue
前面讲的都是数组操作的安全集合，所以ConcurrentLinkedQueue就是链表操作的安全集合。
ConcurrentLinkedQueue 是一个基于链接节点的无界、线程安全的非阻塞队列。它基于 CAS (Compare-And-Swap) 机制实现，这使得它在高并发场景下的性能表现非常出色，尤其是在多核处理器上。
它的核心特点是：
- 高性能：非阻塞，没有锁竞争，吞吐量高。
- 无界：没有容量限制，可以一直添加元素，直到内存耗尽。
- 线程安全：所有公开的队列操作都是原子性的，并且满足
happens-before原则。 - FIFO：严格遵循先进先出原则。
常用方法
队列核心操作方法
| 方法签名 | 功能描述 | 返回值 | 示例/注意 |
|---|---|---|---|
add(E e) | 将指定元素插入此队列的尾部。 由于队列是无界的，这个方法永远不会抛出 IllegalStateException。 | boolean (总是返回 true) | queue.add("Task 1"); |
offer(E e) | 将指定元素插入此队列的尾部。 与 add() 功能相同，但它更符合队列的通用接口规范。 | boolean (总是返回 true) | queue.offer("Task 2"); |
remove() | 获取并移除此队列的头部元素。 如果队列为空，则抛出 NoSuchElementException。 | E (队列头部的元素) | String task = queue.remove(); |
poll() | 获取并移除此队列的头部元素。 如果队列为空，则返回 null。 | E (队列头部的元素) 或 null | String task = queue.poll(); // 更安全，不会抛出异常 |
element() | 获取但不移除此队列的头部元素。 如果队列为空，则抛出 NoSuchElementException。 | E (队列头部的元素) | String firstTask = queue.element(); |
peek() | 获取但不移除此队列的头部元素。 如果队列为空，则返回 null。 | E (队列头部的元素) 或 null | String firstTask = queue.peek(); // 更安全 |
查询方法
| 方法签名 | 功能描述 | 返回值 |
|---|---|---|
size() | 返回此队列中的元素数量。 | int |
isEmpty() | 如果此队列不包含任何元素，则返回 true。 | boolean |
contains(Object o) | 如果此队列包含指定元素，则返回 true。 | boolean |
多的不说直接上源码看下他是如何实现线程安全的
数据结构
public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable {
transient volatile Node<E> head; // 头节点
private transient volatile Node<E> tail; // 尾节点
static final class Node<E> { // 内部类，定义的节点的数据结构，很常规的单向链表结构
volatile E item;
volatile Node<E> next;
}
}
定义数据结构时都使用到了volatile 关键字，保证数据的可见性。
获取元素
public E poll() {
restartFromHead: // 是一个标签，用于在特定情况下跳出内层循环，重新从头部开始遍历
for (;;) { // 外层死循环确保操作最终会成功完成
for (Node<E> h = head, p = h, q;; p = q) { 内层循环初始化：h 记录当前头节点，p 作为遍历指针从 h 开始
final E item;
if ((item = p.item) != null && p.casItem(item, null)) { // 尝试cas操作，将当前节点设置null值，这是一个关键竞争点：CAS 操作保证只有一个线程能成功
if (p != h) // cas成功，但是当前节点不是头节点
updateHead(h, ((q = p.next) != null) ? q : p); // updateHead 将 head 指向新的位置，帮助后续操作跳过已处理的节点
return item;
}
else if ((q = p.next) == null) { // cas失败说明出现的线程间竞争，这个时候如果头节点下一个节点为空，说明队列元素被其他线程拿完了，队列为空，当前线程已经没有元素可以拿了，直接返回null
updateHead(h, p);
return null;
}
else if (p == q) // 如果 p == q，说明节点已被出队（自引用），需要重新开始
continue restartFromHead;
}
}
}
boolean casItem(E cmp, E val) {
return ITEM.compareAndSet(this, cmp, val); // cas操作尝试交换值
}
final void updateHead(Node<E> h, Node<E> p) {
//如果p节点不是头节点，那么就把p节点赋值给头节点，最后释放头节点
if (h != p && HEAD.compareAndSet(this, h, p))
NEXT.setRelease(h, h);
}
上面代码看着脑袋嗡嗡的，用AI写了个例子方便理解：
初始状态
head → [Node1: item="A"] → [Node2: item="B"] → [Node3: item="C"]
时间线：
-
Thread-1 进入
poll()，读取head指向 Node1(“A”) -
Thread-2****几乎同时进入
poll()，也读取head指向 Node1(“A”) -
Thread-1 执行
p.casItem("A", null)成功-
现在：Node1.item = null，这里Thread-2是能感知得到的，因为所有节点都是volatile
-
Thread-1 继续执行，发现
p != h为 false（因为p就是h） -
Thread-1 直接返回 “A”
-
-
此时 Thread-2 的处境：
-
Thread-2 也尝试执行
p.casItem("A", null) -
但是 此时 Node1.item 已经是
null了 -
CAS 操作失败！因为期望值是 “A”，但实际值是
null
-
-
Thread-2 的处理：
// CAS 失败，进入 else 分支 else if ((q = p.next) == null) { // q = Node2, 不为null → 跳过 else if (p == q) // p != q → 跳过 // 执行 p = q，现在 p 指向 Node2("B") -
Thread-2 继续循环：
-
现在
p指向 Node2(“B”) -
尝试
p.casItem("B", null) -
如果此时没有其他线程竞争，Thread-2 成功获取 “B”
-
推进状态流程：
// 竞争前：
[A] → [B] → [C]
// Thread-1 成功，Thread-2 失败后：
[null] → [B] → [C]
↑
head (thread-1更新)
// Thread-2 重试成功：
[null] → [null] → [C]
↑
head (thread-2更新)
自引用检测情况
p == q 的情况处理节点被出队后的清理状态，防止遍历已删除的节点。
这种设计确保了在高并发环境下，多个线程可以同时进行出队操作而不会破坏队列的一致性，每个线程最终都能成功获取元素或确认队列为空。
添加元素
public boolean offer(E e) {
final Node<E> newNode = new Node<E>(Objects.requireNonNull(e)); // 创建新节点
for (Node<E> t = tail, p = t;;) { //记录当前的 tail 到局部变量 t，p 作为查找指针从 t 开始，循环保证元素一定添加成功
Node<E> q = p.next; // q指向尾节点下一个节点，也就是新元素添加的位置
if (q == null) { // 尾节点下一个元素为空说明，可以被挂在尾节点之后
if (NEXT.compareAndSet(p, null, newNode)) { // cas尝试设置新节点
if (p != t) // p != t 说明我们遍历过程中发现了 tail 滞后，说明是第二次循环时才cas成功，第一次cas失败，被其他线程先设置了新值，此时t已经被其他线程更新为新值了
TAIL.weakCompareAndSet(this, t, newNode); // 更新尾节点的位置
return true;
}
}
else if (p == q)
p = (t != (t = tail)) ? t : head; // 遇到自引用节点，需要重新定位
else
p = (p != t && t != (t = tail)) ? t : q; // 推进p指针：如果tail有更新就跳转到新tail，否则正常向后移动
}
}
还是举一个例子：
初始状态：
[Node1] → [Node2] → null
↑ ↑
tail,p q
Thread-A 和 Thread-B 同时执行 offer：
-
Thread-A 读取
t = tail = Node2,p = Node2 -
Thread-B 读取
t = tail = Node2,p = Node2 -
Thread-A 执行：
-
-
q = Node2.next = null -
CAS(Node2.next, null, NodeA)成功，返回true
此时状态变为
[Node1] → [Node2] → [NodeA] → null ↑ ↑ tail,p q // 这个时候thread-A并没有更新tail的位置 -
-
Thread-B 执行（在Thread-A之后）：
-
此时thread-b尝试cas发现，尾节点的下一个节点(nodeA)不为null,cas失败
-
到下一遍循环，此时q = p.next 也就是nodeA, p!=q
-
推进下一个节点，p = q(nodeA)
-
q = p.next = null，尝试cas操作成功
-
p(nodeA),t(node2)两个不相等更新尾节点位置，TAIL.weakCompareAndSet(this, t, newNode)
-
最终状态：
[Node1] → [Node2] → [NodeA] → [NodeB] → null
↑
tail
p==q情况的是考虑到了如果我线程正在入队，但是另外的线程正在读取元素出队，那么就可能导致出现自引用，需要重置尾节点。
ConcurrentLinkedDeque
ConcurrentLinkedDeque 是 Java 并发包中的一个非常重要的类，它提供了一个线程安全、无界、非阻塞的双端队列（Double-Ended Queue，简称 Deque）。
简单来说，你可以把它想象成一个既能在队头添加/删除元素，又能在队尾添加/删除元素的线程安全队列，并且性能极高，因为它内部同样采用了无锁的 CAS (Compare-And-Swap) 机制。
队列操作常用方法
| 方法签名 | 功能描述 | 返回值 | 示例/注意 |
|---|---|---|---|
push(E e) | 将元素推入此双端队列所表示的栈的顶部（即队头）。等效于 addFirst(e)。 如果队列已满（此队列永远不会满）或由于其他原因无法添加，则抛出 IllegalStateException 或 NullPointerException。 | void | deque.push("A"); // 现在A在队头 |
pop() | 弹出此双端队列所表示的栈的顶部元素（即队头）。等效于 removeFirst()。 如果队列为空，则抛出 NoSuchElementException。 | E | String elem = deque.pop(); |
peek() | 获取但不移除此双端队列所表示的栈的顶部元素（即队头）。等效于 peekFirst()。 如果队列为空，则返回 null。 | E 或 null | String top = deque.peek(); |
poll() | 获取并移除此双端队列所表示的栈的顶部元素（即队头）。等效于 pollFirst()。 如果队列为空，则返回 null。 | E 或 null | String elem = deque.poll(); |
add(E e) | 将指定元素添加到此双端队列的队尾。等效于 addLast(e)。 | boolean (总是 true) | deque.add("Z"); // 现在Z在队尾 |
offer(E e) | 将指定元素添加到此双端队列的队尾。等效于 offerLast(e)。 | boolean (总是 true) | deque.offer("Y"); |
remove() | 获取并移除此双端队列的队头元素。等效于 removeFirst()。 | E | String elem = deque.remove(); |
pollFirst() / pollLast() | 分别从队头/队尾获取并移除元素。如果为空，返回 null。 | E 或 null | deque.pollFirst();, deque.pollLast(); |
peekFirst() / peekLast() | 分别查看队头/队尾元素，不移除。如果为空，返回 null。 | E 或 null | deque.peekFirst();, deque.peekLast(); |
元素获取常用方法
| 方法签名 | 功能描述 | 返回值 |
|---|---|---|
addFirst(E e) | 将指定元素插入此双端队列的队头。 | void |
addLast(E e) | 将指定元素插入此双端队列的队尾。 | void |
offerFirst(E e) | 将指定元素插入此双端队列的队头。 | boolean (总是 true) |
offerLast(E e) | 将指定元素插入此双端队列的队尾。 | boolean (总是 true) |
removeFirst() / removeLast() | 从队头/队头移除一个元素。如果为空，抛出 NoSuchElementException。 | E |
getFirst() / getLast() | 获取但不移除队头/队尾元素。如果为空，抛出 NoSuchElementException。 | E |
源码这里就不解释了，底层实现也是采用cas+重试保证线程的安全性，但因为是双向链表，所以实现更为复杂，有兴趣的可以点开源码看看。
锁分段容器
Hashtable
Hashtable 是 Java 集合框架中的一个基础类，它实现了 Map 接口，用于存储键值对。它是 Java 早期（JDK 1.0）就引入的类，以线程安全著称。
它其实和hashMap实现上差不多，线程安全也是每一个方法都使用了synchronized关键字，从而保证了并发安全。
Hashtable 的性能与缺陷
由于全表锁的机制，Hashtable 在现代并发应用中暴露出明显的性能问题。
性能瓶颈（读/写均受影响）：
- 写操作：多个线程同时写 Hashtable 时，它们必须串行执行，等待时间会非常长。
- 读操作：即使是只读的高并发场景，一个线程在执行 get() 方法时，其他所有线程（无论是读还是写）也必须等待，这极大地限制了读取的吞吐量。
- 迭代器的弱一致性：Hashtable 的迭代器是快速失败的。如果在迭代过程中，有其他线程修改了 Hashtable 的结构（通过 put, remove 等），迭代器会立即抛出 ConcurrentModificationException。这使得在遍历 Hashtable 的同时进行并发修改变得非常困难和危险。
ConcurrentHashMap
ConcurrentHashMap 是一个支持高并发、高吞吐量的线程安全 HashMap 实现。它与 Hashtable 和 Collections.synchronizedMap 不同，它使用更细粒度的锁机制来实现更高的并发性能。
JDK7实现线程安全使用的是分段锁机制
// JDK 7 的结构
ConcurrentHashMap
├── Segment[] (16个段)
│ ├── Segment 0 → HashEntry[] → 链表
│ ├── Segment 1 → HashEntry[] → 链表
│ └── ... (每个段独立加锁)
缺点：并发度受限于段的数量（默认16）
JDK8之后就对其改进了，采用更细粒度的cas + synchronized实现
// JDK 8+ 的结构
ConcurrentHashMap
├── Node[] table
│ ├── 索引0 → Node | TreeBin | ReservationNode...
│ ├── 索引1 → 链表 或 红黑树
│ └── ...
这里就讲解一下1.8之后的源码：
数据结构
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable {
// 哈希表，volatile 保证可见性
transient volatile Node<K,V>[] table;
// 下一个要使用的表，仅在扩容时非空
private transient volatile Node<K,V>[] nextTable;
// 基础计数器，主要在没有竞争时使用
private transient volatile long baseCount;
// 表初始化和扩容的控制字段
private transient volatile int sizeCtl;
static class Node<K,V> implements Map.Entry<K,V> { // 链表节点的数据结构
final int hash;
final K key;
volatile V val; // volatile 保证可见性
volatile Node<K,V> next; // volatile 保证可见性
// CAS 更新 value
boolean casVal(V cmp, V val) {
return UNSAFE.compareAndSwapObject(this, valOffset, cmp, val);
}
}
// 红黑树节点数据结构 继承了node节点
static final class TreeNode<K,V> extends Node<K,V> {
TreeNode<K,V> parent;
TreeNode<K,V> left;
TreeNode<K,V> right;
TreeNode<K,V> prev;
boolean red;
}
}
添加元素
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); // 高位和地位进行异或增加hash指的随机性
int binCount = 0; // 记录此次链表长度，是否转为红黑树
// 自旋插入，保证执行成功
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh; K fk; V fv;
// 1. 如果桶为空就初始化桶，里面还有逻辑是，初始化时如果有其他线程正在初始化就yeild当前线程
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)))
break; // cas设置值成功，跳出循环，返回
}
// 3.如果map正在扩容，哪么就帮忙扩容，但是帮忙线程有上限，最多不超过桶的大小
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
// 4. 如果不允许覆盖旧值，直接返回，这里底层默认时false允许覆盖新值
else if (onlyIfAbsent
&& fh == hash
&& ((fk = f.key) == key || (fk != null && key.equals(fk)))
&& (fv = f.val) != null)
return fv;
else {
V oldVal = null;
// 5. 如果头节点有值了，对头节点进行加锁，注意这里是对头节点加锁！！！
synchronized (f) {
//5.1 链表第一个节点等于头节点就是链表，然后将新的节点查到链表尾部
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) { // 这里遍历时还对链表长度计数了binCount++
// 尾插操作...
}
}
//5.2 链表头节点不是头节点那么就是树，就把节点加入到树中
else if (f instanceof TreeBin) {
binCount = 2;
// 走红黑树添加节点逻辑
}
}
}
// 6. 加节点结束后，检查添加是否成功，是否需要树化
if (binCount != 0) {
if (binCount >= 8) // 链表bincount才会大于8，树操作bincount = 2，所以bincount>8尝试对链表进行树化，
treeifyBin(tab, i); // 但是treeifyBin里面还判断了桶容量如果小于64，先扩容，大于64将来链表转为红黑树
if (oldVal != null) // 如果新增的值是覆盖的了旧值的逻辑，就不会涉及到后面的扩容和树化操作了，所以这里就直接返回了
return oldVal;
break;
}
}
}
addCount(1L, binCount); // 检查并触发扩容或者树化
return null;
}
获取元素
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 1. 计算 hash
int h = spread(key.hashCode());
// 2. 表不为空且桶不为空
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 3. 检查头节点
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 4. hash < 0 表示特殊节点（树节点或转发节点）
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
// 5. 遍历链表
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
get方法整个流程都是无锁操作，所以hashMap读的效率是很高的。
为什么没有看到进行树查询的分支？
这里利用到了java的多态，关键点e.find(h,key)，如果此时e类型为TreeBin树的root节点，那么调用的就是树的find方法。
// TreeBin.find()中的读锁机制
if (U.compareAndSetInt(this, LOCKSTATE, s, s + READER)) {
try {
// 执行红黑树查找
p = root.findTreeNode(h, k, null);
} finally {
// 释放读锁
U.getAndAddInt(this, LOCKSTATE, -READER);
}
}
如果这里是链表节点
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}
对比链表查找和树查找可以发现，直接时遍历链表查询，没有加任何锁，因为值时volatile类别的，线程之间是可见的，所以没有加锁。
而进行树查找时进行了cas操作，这里就是对读线程进行计数，如果后面有线程要修改这颗树会被阻塞。
为什么链表结构读取没有加锁而树查询需要加读锁计数器？
因为链表结构简单，而且采用的是尾插法，value又是volatile的,线程之间可见的所以没有加锁。
而红黑树结构复杂，读取过程中如果红黑树结构变化，有可能就读不到值，所有进行树查询时，就会进行读锁计数方式，读线程读取时写线程就不能修改树增加节点，链表则可以读写并发。
使用的几个注意事项
复合操作
// 复合操作不是原子性的
if (!map.containsKey("key")) { // 操作1
map.put("key", 1); // 操作2
}
// 可以使用以下方法避免
map.putIfAbsent(key, value); // 使用 putIfAbsent
map.computeIfAbsent(key, k -> createExpensiveValue(k)); // 使用 computeIfAbsent（）
map.compute(key, (k, v) -> v == null ? 1 : v + 1); // 使用 compute
批量操作问题
// 批量操作不是原子的
public void addAll(Map<String, Integer> other) {
for (Map.Entry<String, Integer> entry : other.entrySet()) {
map.put(entry.getKey(), entry.getValue()); // 不是原子批量操作
}
}
// 使用 putAll（但仍有时间窗口）
map.putAll(otherMap);
// 更安全的方式：在构造时传入
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>(initialMap);
迭代器弱一致性问题
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
map.put("a", "1");
map.put("b", "2");
Iterator<String> it = map.values().iterator();
// 在迭代过程中修改
map.put("c", "3");
map.remove("a");
while (it.hasNext()) {
System.out.println(it.next());
// 可能输出: 1, 2 或 2 或 1, 2, 3
// 不会抛出 ConcurrentModificationException
// 但可能看到部分更新，也可能看不到
}
原子类
都是无锁安全的原子类，基于CAS + volatitle + 自旋实现。
列如AtomicInteger增加操作底层实现：
private static final Unsafe U = Unsafe.getUnsafe();
private static final long VALUE = U.objectFieldOffset(AtomicInteger.class, "value");
private volatile int value;
publi final int getAndAddInt(Object o, long offset, int delta) {
int v;
// 经典的 CAS 自旋循环！
do {
// 1. 首先，获取当前最新的值，作为 CAS 操作中的“预期原值”
v = getIntVolatile(o, offset); // 底层是 volatile 读，保证拿到最新值
// 2. 尝试进行 CAS 操作：如果当前值还是 v，就把它设置为 v + delta
// 3. 如果 CAS 失败（当前值不等于 v 了，说明被其他线程改了），
// 则循环重试，直到成功为止。
} while (!compareAndSwapInt(o, offset, v, v + delta));
// 循环退出后，返回修改之前的值 (v)
return v;
}
基本原子类
AtomicInteger：整型原子类AtomicBoolean：布尔型原子类AtomicLong：长整型原子类
public final int get() //获取当前的值
public final int getAndSet(int newValue)//获取当前的值，并设置新的值
public final int getAndIncrement()//获取当前的值，并自增
public final int getAndDecrement() //获取当前的值，并自减
public final int getAndAdd(int delta) //获取当前的值，并加上预期的值
boolean compareAndSet(int expect, int update) //如果输入的数值等于预期值，则以原子方式将该值设置为输入值（update）
public final void lazySet(int newValue)//最终设置为newValue,使用 lazySet 设置之后可能导致其他线程在之后的一小段时间内还是可以读到旧的值。
数组原子类
AtomicIntegerArray：整型数组原子类AtomicLongrArray：长整型数组原子类AtomicReferenceArray：用类型数组原子类
public final int get(int i) //获取 index=i 位置元素的值
public final int getAndSet(int i, int newValue)//返回 index=i 位置的当前的值，并将其设置为新值：newValue
public final int getAndIncrement(int i)//获取 index=i 位置元素的值，并让该位置的元素自增
public final int getAndDecrement(int i) //获取 index=i 位置元素的值，并让该位置的元素自减
public final int getAndAdd(int i, int delta) //获取 index=i 位置元素的值，并加上预期的值
boolean compareAndSet(int i, int expect, int update) //如果输入的数值等于预期值，则以原子方式将 index=i 位置的元素值设置为输入值（update）
public final void lazySet(int i, int newValue)//最终 将index=i 位置的元素设置为newValue,使用 lazySet 设置之后可能导致其他线程在之后的一小段时间内还是可以读到旧的值。
原子引用类
AtomicReference:引用类型原子类AtomicStampedReference：原子更新带有版本号的引用类型。该类将整数值与引用关联起来，可用于解决原子的更新数据和数据的版本号，可以解决使用 CAS 进行原子更新时可能出现的 ABA 问题。解决修改过几次AtomicMarkableReference：原子更新带有标记的引用类型。该类将 boolean 标记与引用关联起来
解决是否修改过，它的定义就是将标记戳简化为true/false，类似于一次性筷子
对象的属性修改原子类
AtomicIntegerFieldUpdater：原子更新对象中int类型字段的值AtomicLongFieldUpdater：原子更新对象中Long类型字段的值AtomicReferenceFieldUpdater：原子更新对象中引用类型字段的值
原子增强类
-
DoubleAccumulator：一个或多个变量，它们一起保持运行double使用所提供的功能更新值 -
DoubleAdder：一个或多个变量一起保持初始为零double总和 -
LongAccumulator：一个或多个变量，一起保持使用提供的功能更新运行的值long ，提供了自定义的函数操作 -
LongAdder：一个或多个变量一起维持初始为零long总和（重点），只能用来计算加法，且从0开始计算这里再高并发下为什么LongAdder会比AutomicLon快很多？
因为LongAdder减小线程之间的CAS操作，内部有一个cell[]数据，可以充分利用cpu核心数进行数据运算，列如系统cpu是8核的，automicLong只利用了一个核心进行计算，Longadder利用8个核心。
LongAdder原理
cas+base+cell的方案，再低并发下没有cas失败的情况下，只进行base的cas操作。当出现cas失败的情况，就新增cell数据，将线程的操作通过hash取余映射到一个cell数组元素中进行数据的加1，减少cas冲突，
最后需要知道最终结果，直接把数据中的数据+bas全部相加就是最终结果。
cell[]可以无限扩大吗？
并不是，longadder核心思想是充分利用cpu计算单元，当cell数据等于cpu核心数时就不会再增加了。所以longadder的效率还和系统核心数有关。
常用阻塞队列
LinkedBlockingQueue
阻塞队列，底层使用链表实现，如果初始化时没有指定容量，那么默认最大容量为Integer.MAX_VALUE，它是最常用的阻塞队列，很多线程池默认以它作为工作队列。
常用方法
不阻塞方法
| 方法 | 说明 | 失败时行为 |
|---|---|---|
| boolean add(E e) | 将元素添加到链表尾部，如果队列已满，抛出 IllegalStateException | 抛出异常 |
| boolean offer(E e) | 将元素添加到队列尾部。如果队列已满，返回 `false` | 返回 `false` |
| boolean offer(E e, long timeout, TimeUnit unit) | 尝试将元素添加到队列尾部，如果在指定时间内还不能添加（队列已满），则放弃。 | 超时后返回 `false` |
| E remove() | 移除并返回队列头部的元素。如果队列为空，抛出 `NoSuchElementException` | 抛出异常 |
| E poll() | 移除并返回队列头部的元素。如果队列为空，返回 `null` | 返回 `null` |
| E peek() | 不移除地获取队列头部的元素。如果队列为空，返回 `null` | 返回 `null` |
阻塞线程方法：
| 方法 | 说明 |
|---|---|
| void put(E e) | 将元素添加到队列尾部。如果队列已满，当前线程会阻塞，直到队列有空余空间。 |
| E take() | 移除并返回队列头部的元素。如果队列为空，当前线程会阻塞，直到队列中有新元素。 |
| int remainingCapacity() | 返回队列当前可用的容量。对于 LinkedBlockingQueue，就是 capacity - size() |
作为线程安全容器使用：
public class Main {
public static void main(String[] args) throws InterruptedException {
// 创建有界队列，容量为10
BlockingQueue<String> queue = new LinkedBlockingQueue<>(10);
// 创建多个生产者线程
for (int i = 0; i < 3; i++) {
final int producerId = i;
new Thread(() -> produce(queue, producerId), "Producer-" + i).start();
}
// 创建多个消费者线程
for (int i = 0; i < 2; i++) {
final int consumerId = i;
new Thread(() -> consume(queue, consumerId), "Consumer-" + i).start();
}
}
// 生产者方法
private static void produce(BlockingQueue<String> queue, int producerId) {
try {
for (int i = 0; i < 5; i++) {
String item = "Item-" + producerId + "-" + i;
// 方法1: put - 阻塞直到空间可用
queue.put(item);
System.out.println(Thread.currentThread().getName() + " 生产: " + item);
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
// 消费者方法
private static void consume(BlockingQueue<String> queue, int consumerId) {
try {
while (true) {
// 方法1: take - 阻塞直到元素可用
String item = queue.take();
System.out.println(Thread.currentThread().getName() + " 消费: " + item);
Thread.sleep(150);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
为什么LinkedBlockingQueue是线程安全的？原理分析
LinkedBlockingQueue结构：
public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>{
private final int capacity; // 队列的容量，如果不传则默认Integer.MAX_VALUE
private final AtomicInteger count = new AtomicInteger(); // 当前队列中元素数量,原子类保证并发安全
transient Node<E> head; // 链表的头指针，head.item = null
private transient Node<E> last; // 链表的尾指针 last.next = null
private final ReentrantLock takeLock = new ReentrantLock(); // 出队操作锁，获取元素的线程公用这把锁
private final Condition notEmpty = takeLock.newCondition(); // 出队条件：非空队列，如果线程获取元素时队列为空，就会加入到notEmpty条件中等待获取元素
private final ReentrantLock putLock = new ReentrantLock(); // 入队操作锁，添加元素的线程公用这把锁
private final Condition notFull = putLock.newCondition(); // 入队条件：非满队列，如果线程添加元素时队列为满了，就会加入到notFull条件中等待添加元素
// 默认构造方法，队列容量为Integer.MAX_VALUE
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
// 指定队列容量的构造方法
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
// 初始化链表的头尾节点
last = head = new Node<E>(null);
}
static class Node<E> {
E item;
Node<E> next;
Node(E x) { item = x; }
}
}
底层其实就是利用了ReentrantLock对入队和出队操作加锁保证容器的安全，使用两个锁和两个等待条件，来实现入队和出队的读写分离，通过构造函数可以看出，如果我们不指定队列的容量，那么
它的最大容量是Integer.MAX_VALUE，通过Node节点可以看出LinkedBlockingQueue是单链表结构，所以添加元素都是添加到链表尾部，获取元素都是获取链表头部元素。
非阻塞添加元素add/offer：
public boolean add(E e) {
if (offer(e)) // add方法其实里面调用的是offer方法
return true;
else
throw new IllegalStateException("Queue full");
}
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity) // 队列元素到达最大值直接返回false
return false;
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock; // 使用写锁
putLock.lock();
try {
if (count.get() < capacity) { // 二次判断防止虽然第一次判断队列没有满，但是获取锁是第二个线程先获取到锁，添加了元素后，队列满了，当前线程阻塞后被唤醒再进行判断满没，满了直接释放锁返回
enqueue(node); // 队列容量足够则入队
c = count.getAndIncrement(); // 队列元素计数 + 1
if (c + 1 < capacity) // 如果队列没有满，就唤醒添加等待队列中的线程
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty(); // 如果之前元素为空，那么肯定有等待读线程，加入元素后唤醒一个读线程
return c >= 0;
}
// 等待一定时间添加元素
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
final int c;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly(); // 获取锁并且可以响应中断
try {
while (count.get() == capacity) { // 队列已满，加入到读等待队列中等待
if (nanos <= 0L)
return false; // 等待时间都完了，还没有添加到元素，直接返回了，不加了
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e)); // 等待时间内不断判断队列容量是否可以满足添加，可以添加则入队
c = count.getAndIncrement(); // 队列数量计数+1
if (c + 1 < capacity)
notFull.signal(); // 如果容量未满，则唤醒因为队满阻塞的写线程
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty(); // 如果之前元素为空，那么肯定有等待读线程，加入元素后唤醒一个读线程
return true;
}
阻塞添加元素：
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
final int c;
final Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly(); // 可终端获取锁
try {
while (count.get() == capacity) { // 队列满了直接加入到写等待队列中
notFull.await();
}
enqueue(node); // 被其他线程唤醒后，添加元素
c = count.getAndIncrement(); // 计数+1
if (c + 1 < capacity)
notFull.signal(); // 队列未满，唤醒其他写线程
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty(); // 如果之前元素为空，那么肯定有等待读线程，加入元素后唤醒一个读线程
}
和前面方法不一样的地方是，put没有先去哦判断队列是否已满，前面是满了直接返回或者抛异常，没有满才去获取锁添加元素，put方法是满了我直接阻塞线程，反正就是这个元素一定要添加，其他的就是加不了就算了。
非阻塞获取方法remove/poll：
public E remove() {
E x = poll(); // 也是调用的poll方法
if (x != null)
return x;
else
throw new NoSuchElementException();
}
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0) // 没有元素返回null
return null;
final E x;
final int c;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock(); // 使用读锁加锁
try {
if (count.get() == 0) // 第二次判断是防止，虽然第一次判断出队列中有元素，但是其他线程先获取到锁，自己等待后被唤醒，发现队列中元素已经被其他线程获取完了，就直接返回
return null;
x = dequeue(); // 队头元素出队
c = count.getAndDecrement(); // 计数-1
if (c > 1)
notEmpty.signal(); // 队列不为空，唤醒其他阻塞的读线程
} finally {
takeLock.unlock();
}
if (c == capacity) // 队列容量之前是满的，现在获取一个元素后会删除，队列就不满了，就可以唤醒之前阻塞的写线程
signalNotFull();
return x;
}
// 超时获取元素
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
final E x;
final int c;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly(); // 可中断获取锁
try {
while (count.get() == 0) { // 队列为空 等待一段时间
if (nanos <= 0L) // 等待时间都到了还没有就直接返回了
return null;
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();
c = count.getAndDecrement(); // 获取到元素后计数-1
if (c > 1)
notEmpty.signal(); // 还有元素唤醒其他被阻塞的读线程
} finally {
takeLock.unlock();
}
if (c == capacity) // 队列容量之前是满的，现在获取一个元素后会删除，队列就不满了，就可以唤醒之前阻塞的写线程
signalNotFull();
return x;
}
阻塞获取方法：
public E take() throws InterruptedException {
final E x;
final int c;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) { // 队列为空，阻塞等待
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement(); // 获取到元素，队列计数-1
if (c > 1)
notEmpty.signal(); // 还有元素唤醒其他被阻塞的读线程
} finally {
takeLock.unlock();
}
if (c == capacity) // 队列容量之前是满的，现在获取一个元素后会删除，队列就不满了，就可以唤醒之前阻塞的写线程
signalNotFull();
return x;
}
和poll/remove不同的是，take是一定要让这个线程拿到元素才会返回，没有就阻塞当前线程，直到有元素获取后才返回，poll是没有我就不获取了，直接返回。
LinkedBlockingDeque
双端阻塞队列，和LinkedBlokingQueue区别在与，LinkedBlokingQueue底层是单链表实现遵循FIFO顺序，LinkedBlockingDeque底层是双向链表，支持双向添加获取元素，所以在底层实现上也有不同，但都是通过ReentrantLock加锁去实现并发安全。不同的是，LinkedBlockingDeque是单锁设计，像LinkedBlockingQueque入队操作链表尾部，出队操作链表头部，实现起来就比较简单，但是LinkedBlockingDeque支持队头队尾操作，两把锁实现困难，所以采用单锁设计。
常用方法
| 场景 | 头部操作 | 尾部操作 | 标准队列操作 |
|---|---|---|---|
| 添加 (阻塞,必须成功) | putFirst(e) | putLast(e) | put(e) |
| 添加 (阻塞,可超时) | offerFirst(e, timeout, unit) | offerLast(e, timeout, unit) | offer(e, timeout, unit) |
| 添加 (非阻塞,失败不关心) | offerFirst(e) | offerLast(e) | offer(e) |
| 添加 (非阻塞,失败抛异常) | addFirst(e) | addLast(e) | add(e) |
| 获取 (阻塞,必须成功) | takeFirst() | takeLast() | take() |
| 获取 (阻塞,可超时) | pollFirst(timeout, unit) | pollLast(timeout, unit) | poll(timeout, unit) |
| 获取 (非阻塞,失败返回null) | pollFirst() | pollLast() | poll() |
| 获取 (非阻塞,失败抛异常) | removeFirst() | removeLast() | remove() |
| 查看 (不移除,失败返回null) | peekFirst() | peekLast() | peek() |
| 查看 (不移除,失败抛异常) | getFirst() | getLast() | - |
原理分析
定义结构
public class LinkedBlockingDeque<E> extends AbstractQueue<E>，implements BlockingDeque<E>{
// 内部节点类
static final class Node<E> {
E item; // 节点存储的元素
Node<E> prev; // 前驱节点
Node<E> next; // 后继节点
Node(E x) {
item = x;
}
}
transient Node<E> first; // 头节点
transient Node<E> last; // 尾节点
private transient int count; // 队列元素计数
private final int capacity; // 队列容量
final ReentrantLock lock = new ReentrantLock(); // 全局锁 - 与 LinkedBlockingQueue 不同，这里只有一个锁
private final Condition notEmpty = lock.newCondition(); // 出队条件变量
private final Condition notFull = lock.newCondition(); // 入队条件变量
}
LinkedBlockingDeque底层采用双向链表，使用同一把锁完成并发安全控制，出队入队条件不同以实现不同的读写阻塞效果。
核心入队方法：
入队各种变种方法最终将元素加到链表中底层都是调用的这个两个方法。
// 链接到队列头部
private boolean linkFirst(Node<E> node) {
if (count >= capacity) // 队列已满返回
return false;
Node<E> f = first;
node.next = f;
first = node;
if (last == null)
last = node;
else
f.prev = node; // 添加元素到
++count; // 计数+1
notEmpty.signal(); // 唤醒阻塞中的读线程
return true;
}
// 链接到队列尾部
private boolean linkLast(Node<E> node) {
if (count >= capacity) // 队列已满返回
return false;
Node<E> l = last;
node.prev = l;
last = node;
if (first == null)
first = node;
else
l.next = node; // 添加元素到链表尾部
++count; // 计数+1
notEmpty.signal(); // 唤醒阻塞中的读线程
return true;
}
非阻塞式入队：
public boolean offerLast(E e) {
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock(); // 获取锁
try {
return linkLast(node); 添加元素到链表尾部
} finally {
lock.unlock();
}
} // 其他非阻塞式获取方法类似的
阻塞式入队：
public void putLast(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock(); // 获得锁
try {
while (!linkLast(node)) // 如果加入链表失败就阻塞线程
notFull.await();
} finally {
lock.unlock();
}
}
阻塞式和非阻塞式区别就在于，阻塞式一定要将元素添加到链表中，非阻塞式则是如果加入不到队列中就返回，处理逻辑由外部程序决定。
核心出队方法：
// 链接到链表头部
private E unlinkFirst() {
Node<E> f = first;
if (f == null)
return null;
Node<E> n = f.next;
E item = f.item;
f.item = null;
f.next = f; // help GC
first = n;
if (n == null)
last = null;
else
n.prev = null; // 元素出队，节点指向指为空方面出队节点GC
--count; // 计数-1
notFull.signal(); // 唤醒阻塞的写线程
return item;
}
private E unlinkLast() { // 和上面逻辑一样，只是这个方法是加载链表尾部
Node<E> l = last;
if (l == null)
return null;
Node<E> p = l.prev;
E item = l.item;
l.item = null;
l.prev = l; // help GC
last = p;
if (p == null)
first = null;
else
p.next = null;
--count;
notFull.signal();
return item;
}
非阻塞式出队：
public E pollFirst() {
final ReentrantLock lock = this.lock;
lock.lock(); // 很简单的源码，配合上面核心方法一看就懂
try {
return unlinkFirst();
} finally {
lock.unlock();
}
}
阻塞式出队：
public E takeFirst() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E x;
while ( (x = unlinkFirst()) == null)
notEmpty.await();
return x;
} finally {
lock.unlock();
}
}
ArrayBlockingQueue
ArrayBlockingQueue是一个有界阻塞队列，底层一个Object数组 + RentrantLocks，通过维护读下标和写下标，完成元素的快速写入和快速读取，整个写入和读取都是循环的， 写到数据尾部写下标就会指向数组第一位，以此循环的利用数组，读操作同样逻辑。支持公平锁和非公平锁的阻塞实现，不指定时RentrantLock默认为非公平锁实现。
常用方法
| 场景 | 添加 (生产者) | 获取/移除 (消费者) | 查看 |
|---|---|---|---|
| 阻塞,必须成功 | put(e) | take() | - |
| 阻塞,可超时 | offer(e, timeout, unit) | poll(timeout, unit) | - |
| 非阻塞,失败返回null | offer(e) | poll() | peek() |
| 非阻塞,失败抛异常 | add(e) | remove() | element() |
源码分析
数据结构:
public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>{
// 核心数据结构：定长数组
final Object[] items;
// 重要的指针和状态变量
int takeIndex; // 下一个要取元素的位置
int putIndex; // 下一个要放元素的位置
int count; // 队列中元素的数量
// 同步控制
final ReentrantLock lock; // 主锁
private final Condition notEmpty; // 非空条件
private final Condition notFull; // 非满条件
public ArrayBlockingQueue(int capacity) {
this(capacity, false); // 如果不指定默认是非公平锁
}
// 可选的公平性策略
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair); // 公平或非公平锁
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
}
通过构造函数可以看到，如果我们不指定公平锁，那么默认使用的就是非公平锁。
入队核心方法：
private void enqueue(E e) {
final Object[] items = this.items;
items[putIndex] = e; // 新增元素到指定添加下标位置
if (++putIndex == items.length) putIndex = 0; // 如果添加下标到了数组末尾，则添加下标又到元素开头，一个循环的过程
count++; // 计数+1
notEmpty.signal(); // 唤醒被阻塞的读线程
}
// 阻塞式入队 - put
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 可中断加锁
try {
while (count == items.length) // 队列已满
notFull.await(); // 等待非满条件
enqueue(e); // 执行入队
} finally {
lock.unlock();
}
}
// 非阻塞式入队 - offer
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length) // 队列已满立即返回
return false;
else {
enqueue(e); // 执行入队
return true;
}
} finally {
lock.unlock();
}
}
// 超时式入队 - offer with timeout
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
checkNotNull(e);
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length) {
if (nanos <= 0) // 超时检查
return false;
nanos = notFull.awaitNanos(nanos); // 等待指定时间
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
出队核心方法：
// 主要的出队方法
private E dequeue() {
final Object[] items = this.items;
E x = (E) items[takeIndex]; // 获取元素
items[takeIndex] = null; // 值为空代表出队元素
// 循环数组处理
if (++takeIndex == items.length)
takeIndex = 0;
count--; // 元素计数减少
// 维护迭代器状态（如果存在迭代器）
if (itrs != null)
itrs.elementDequeued();
notFull.signal(); // 唤醒等待的生产者
return x;
}
// 阻塞式出队 - take
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) // 队列为空
notEmpty.await(); // 等待非空条件
return dequeue(); // 执行出队
} finally {
lock.unlock();
}
}
// 非阻塞式出队 - poll
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue(); // 队列空返回null
} finally {
lock.unlock();
}
}
// 超时式出队 - poll with timeout
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
if (nanos <= 0) // 超时检查
return null;
nanos = notEmpty.awaitNanos(nanos); // 等待指定时间
}
return dequeue();
} finally {
lock.unlock();
}
}
优先级队列PriorityBlockingQueue
优先级阻塞队列，可以实现任务的优先级顺序执行，常用于复杂的任务调度，底层是大小堆（数组实现） + ReentrantLock实现并发安全，PriorityBlockingQueue是无界队列，自动扩容的，所以容量没有上限。
常用方法
| 场景 | 添加 (生产者) | 获取/移除 (消费者，按优先级) | 查看 (按优先级) |
|---|---|---|---|
| 阻塞,必须成功 | put(e) (但不阻塞) | take() | - |
| 阻塞,可超时 | - (无入队超时) | poll(timeout, unit) | - |
| 非阻塞,失败返回null | offer(e) (但不阻塞) | poll() | peek() |
| 非阻塞,失败抛异常 | add(e) (但不阻塞) | remove() | element() |
适用场景：
- 任务优先级调度器: 这是它的核心用途。例如，在一个线程池中，系统需要优先处理高优先级的任务，而低优先级的任务可以稍后或甚至被丢弃。
- 事件驱动系统: 在游戏或网络服务中，不同的事件（如用户输入、AI 行为、网络消息）可能具有不同的优先级。可以用它来保证关键事件（如玩家死亡、网络断开）能优先被处理。
- Dijkstra 算法等图论算法: 这些算法需要一个优先级队列来选择下一个要处理的节点，PriorityBlockingQueue 可以在多线程环境下实现这些算法。
- 缓冲区与批处理: 作为高优先级事件和低优先级事件之间的缓冲区。例如，一个生产者不断地产生各种优先级的日志事件，几个高优先级的消费者线程只处理 ERROR 级别的日志，而另一个消费者线程定期（或当队列不为空时）批量处理其他级别的日志
PriorityBlockingQueue最重要的功能就是对元素排序，所以队列中的元素都是必须实现Comparable接口的元素，排序就根据compareTo规则进行优先级的排序：
1.无参构造方法
PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>(); // 其初始默认容量为 11
2.指定容量
PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>(50);
注意@！！！，此时队列还是无界的只是指定了初始容量，后面如果超过指定容量会自动扩容
3.PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator)
功能: 创建一个指定初始容量和自定义 Comparator 的 PriorityBlockingQueue。
用途: 当元素的类型无法修改（例如，使用第三方库的类，且不想让它实现 Comparable），或者有复杂的、非标准的排序规则时非常有用。即使元素本身实现了 Comparable，这个 Comparator 也会覆盖它的自然顺序。
列如：
Comparator<Task> comparator = Comparator.comparingInt(Task::getPriority).reversed();
PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>(11, comparator);
4.PriorityBlockingQueue(Collection<? extends E> c)
功能: 创建一个包含指定集合元素的 PriorityBlockingQueue。
注意: 这会一次性将集合中的所有元素构造成一个有效的堆。如果集合中的元素不满足排序规则（无法比较），构造过程会立即失败。
使用示例：资源分配场景
public class ResourceManager {
private final PriorityBlockingQueue<ResourceRequest> requestQueue =
new PriorityBlockingQueue<>(50,
Comparator.comparing(ResourceRequest::getUrgency)
.thenComparing(ResourceRequest::getTimestamp));
public void requestResource(ResourceRequest request) {
requestQueue.put(request);
System.out.println("资源请求已加入队列: " + request);
}
public void allocateResources() {
new Thread(() -> {
while (true) {
try {
ResourceRequest request = requestQueue.take();
if (canAllocate(request)) {
allocate(request);
} else {
// 无法立即分配，重新放入队列
requestQueue.put(request);
Thread.sleep(1000); // 等待资源释放
}
} catch (InterruptedException e) {
break;
}
}
}).start();
}
}
源码分析
数据结构：
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
// 底层数据结构：平衡二叉堆（数组实现）
private transient Object[] queue;
// 元素数量
private transient int size;
// 比较器，为null时使用自然顺序
private transient Comparator<? super E> comparator;
// 同步控制
private final ReentrantLock lock = new ReentrantLock();
private final Condition notEmpty = lock.newCondition();
// 扩容时的自旋锁
private transient volatile int allocationSpinLock;
// 序列化支持
private PriorityQueue<E> q;
}
堆操作核心方法:
大小堆每次加入一个元素时，根据优先级，堆低的节点通过上浮操作到堆顶，堆顶的节点通过下沉操作到堆低，这样每次获取获取堆顶的元素就是优先级高的或者最低的。
// 上浮操作 - 插入元素时维持堆性质
private void siftUp(int k, E x) {
if (comparator != null)
siftUpUsingComparator(k, x, queue, comparator);
else
siftUpComparable(k, x, queue);
}
// 使用比较器的上浮
private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
Comparator<? super T> cmp) {
while (k > 0) {
int parent = (k - 1) >>> 1; // 父节点索引
Object e = array[parent];
if (cmp.compare(x, (T) e) >= 0) // 如果大于等于父节点，停止上浮
break;
array[k] = e; // 父节点下移
k = parent;
}
array[k] = x; // 找到合适位置
}
// 下沉操作 - 删除元素时维持堆性质
private void siftDown(int k, E x) {
if (comparator != null)
siftDownUsingComparator(k, x, queue, size, comparator);
else
siftDownComparable(k, x, queue, size);
}
// 使用比较器的下沉
private static <T> void siftDownUsingComparator(int k, T x, Object[] array,
int n, Comparator<? super T> cmp) {
int half = n >>> 1; // 最后一个非叶子节点
while (k < half) {
int child = (k << 1) + 1; // 左子节点
Object c = array[child];
int right = child + 1; // 右子节点
// 选择较小的子节点
if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
c = array[child = right];
// 如果x小于等于子节点，停止下沉
if (cmp.compare(x, (T) c) <= 0)
break;
array[k] = c; // 子节点上移
k = child;
}
array[k] = x; // 找到合适位置
}
扩容机制：
底层是数组，所以需要进行扩容操作，当容量小于64时每次容量增加一倍，容量大于64时每次增加1/2。
// 扩容方法
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // 扩容时释放主锁，允许其他操作继续
Object[] newArray = null;
// 使用CAS控制扩容，防止多个线程同时扩容
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
try {
// 计算新容量：小队列翻倍，大队列增长50%
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) : // 小容量增长快
(oldCap >> 1)); // 大容量增长慢
// 处理溢出
if (newCap - MAX_ARRAY_SIZE > 0) {
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0; // 释放自旋锁
}
}
// 如果当前线程没有获得扩容权，让出CPU
if (newArray == null)
Thread.yield();
lock.lock(); // 重新获取锁
// 执行数组拷贝
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
入队和出队操作：
// 入队操作
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock(); // 获取锁
int n, cap;
Object[] array;
// 需要扩容的情况
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null) // 没有比较器，默认使用元素默认实现的比较规则
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp); // 指定了自定义比较器则是使用初始化时指定的
size = n + 1;
notEmpty.signal(); // 唤醒等待的消费者
} finally {
lock.unlock();
}
return true;
}
// 出队操作
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
int n = size - 1;
if (n < 0)
return null; // 队列为空
Object[] array = queue;
E result = (E) array[0]; // 堆顶元素（最小元素）
E x = (E) array[n]; // 最后一个元素
array[n] = null; // 帮助GC
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftDownComparable(0, x, array, n);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
return result;
}
// 阻塞式出队
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
while ((result = dequeue()) == null)
notEmpty.await(); // 队列空时等待
} finally {
lock.unlock();
}
return result;
}
延迟队列DelayQueue
DelayQueue是一个无界阻塞队列，其中的元素必须实现Delayed接口。只有在元素的延迟时间到达时，才能从队列中取出元素。底层实现是一个优先级队列PriorityQueue + ReentrantLock实现。
需要注意的是，队列中的元素必须实现Delayed接口，应为排序依据就是根据等待时间去排序的。它和PriorityBlockingQueue区别在于，DelayQueue是根据延迟时间进行排序（不是绝对也可以自定义实现），并且直接使用的PriorityQueue，PriorityBlockingQueue自定义比较逻辑，并且内部自己实现优先级队列。
常用方法
| 操作 | 抛出异常 | 返回特定值 | 一直阻塞 | 阻塞指定时间 |
|---|---|---|---|---|
| 放数据 | add() | offer() | put() | offer(e, time, unit) |
| 取数据（同时删除数据） | remove() | poll() | take() | poll(time, unit) |
| 取数据（不删除） | element() | peek() | 不支持 | 不支持 |
方法区别：
- 当队列为空的时候，再次取数据，remove()会抛出异常，poll()会返回null，take()会一直阻塞，poll(time, unit)会阻塞指定时间，然后返回null。
使用场景
- 金融系统-订单的超时取消
- 定时任务调度
- 缓存系统 - 实现TTL过期机制
- 消息队列 - 延迟消息处理
原理分析
数据结构：
public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> {
private final transient ReentrantLock lock = new ReentrantLock(); // 全局锁
private final PriorityQueue<E> q = new PriorityQueue<E>(); // 优先级队列，实现元素的排序
private Thread leader = null; //指定线程等待队列头部元素，领导者-跟随者模式
private final Condition available = lock.newCondition(); // 可获取条件
}
添加元素：
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e); // 添加元素到优先级队列中
if (q.peek() == e) { // 堆顶元素是当前元素，领导者线程变更，唤醒等待中的线程
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
阻塞获取元素：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 可中断获取锁
try {
for (;;) {
E first = q.peek(); // 获取堆顶元素
if (first == null) // 堆为空，线程阻塞等待被唤醒
available.await();
else { // 堆顶元素不为空
long delay = first.getDelay(NANOSECONDS); // 获取剩余时间
if (delay <= 0L)
return q.poll(); // 剩余时间<0说明到点了，返回该元素
first = null;
if (leader != null) // 领导者线程不为空，说明当前线程不是第一个来获取的线程，直接等待阻塞直到被唤醒
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread; // 将当前线程设置为领导者线程，并且超时等待剩余时间自动唤醒
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread) // 当前线程是领导者线程则它超时等待自动唤醒后，再次获取前需要将leader置为空，不然下一个循环的这一步前面判断就会判断当前线程不是领导者线程
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null) // 领导者线程获取到堆顶元素后，如果堆还有其他元素，唤醒其他等待的线程成为新的领导者线程
available.signal();
lock.unlock();
}
}
领导者模式优势：
- 减少不必要的线程唤醒
- 只有一个线程在等待特定的延迟时间
- 其他线程无限期等待，直到被唤醒
非阻塞获取元素：
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock(); // 加锁
try {
E first = q.peek();
return (first == null || first.getDelay(NANOSECONDS) > 0) // 堆顶元素为空或者还没有到时间都返回为空，否则返回堆顶元素
? null
: q.poll();
} finally {
lock.unlock();
}
}
public E remove() {
E x = poll();
if (x != null)
return x;
else // 堆顶元素获取失败抛出异常
throw new NoSuchElementException();
}
同步队列SynchronousQueue
SynchronousQueue是一个不存储元素的阻塞队列，每个put操作必须等待一个take操作，反之亦然。它实现了直接的线程间数据传递。它的实现使用的锁并不是ReentrantLock而是使用Locksupport自己实现的等待阻塞机制。
特点：
- 容量为0，不存储任何元素
- put和take必须成对出现
- 支持公平和非公平两种模
公平和非公平的区别：
- 公平：获取元素时严格按照等待时间获取元素
- 非公平：后到的线程也可能先获取到元素。
阻塞队列对比
| 特性 | LinkedBlockingQueue | LinkedBlockingDeque | ArrayBlockingQueue | PriorityBlockingQueue | DelayQueue | SynchronousQueue |
|---|---|---|---|---|---|---|
| 数据结构 | 单向链表 | 双向链表 | 数组 | 二叉堆(数组) | 优先级堆 | 无存储 |
| 边界 | 可选有界(默认Integer.MAX_VALUE) | 可选有界(默认Integer.MAX_VALUE) | 固定有界 | 无界（自动扩容） | 无界（自动扩容） | 容量为0 |
| 锁机制 | 两把锁(putLock/takeLock) | 一把锁 | 一把锁 | 一把锁 | 一把锁 | 无锁(CAS) |
| 公平性 | 不支持 | 不支持 | 可选 | 不支持 | 不支持 | 可选 |
| 阻塞特性 | 队列满/空时阻塞 | 队列满/空时阻塞 | 队列满/空时阻塞 | 队列空时阻塞 | 延迟未到阻塞 | 必须配对阻塞 |
| 排序 | FIFO | FIFO/LIFO | FIFO | 按优先级 | 按延迟时间 | 无 |
参考文章
更多推荐
4
0- 0
已为社区贡献1条内容
所有评论(0)