java队列实现原理

Java 队列实现原理 ## 一、队列的本质 队列是一种先进先出（FIFO） 的线性数据结构，核心操作只有两个： 入队（enqueue）→ [ 元素A | 元素B | 元素C ] → 出队（dequeue） front rear ## 二、Java Queue 体系总览 Iterable └── Collection └── Queue (接口) ├── Deque (双端队列接口) │ ├── ArrayDeque ← 数组实现 │ └── LinkedList ← 双向链表实现 ├── PriorityQueue ← 堆（数组实现） ├── ArrayBlockingQueue ← 环形数组 + 锁 ├── LinkedBlockingQueue ← 链表 + 双锁 └── ConcurrentLinkedQueue ← CAS 无锁链表 ## 三、核心实现原理逐一拆解 ### 1. ArrayDeque — 数组循环队列 这是最常用的非阻塞队列实现，基于循环数组。 java public class ArrayDeque<E> extends AbstractCollection<E> implements Deque<E> { transient Object[] elements; // 底层数组 transient int head; // 队头指针 transient int tail; // 队尾指针 } 关键原理 — 循环利用数组空间： 数组容量 = 8 状态1: head=0, tail=4 [A][B][C][D][ ][ ][ ][ ] ↑h ↑t 状态2: 出队A、B后，head=2 [ ][ ][C][D][ ][ ][ ][ ] ↑h ↑t 状态3: 继续入队 E,F,G,H,I → 数组扩容 [I][ ][C][D][E][F][G][H] ↑t ↑h tail 绕回到数组头部，利用前面空出的位置！ 指针移动的核心代码： java // 入队 — 尾部插入 public boolean offer(E e) { elements[tail] = e; // 关键：位运算取模，tail 到末尾后绕回 0 if ((tail = (tail + 1) & (elements.length - 1)) == head) doubleCapacity(); // tail 追上 head → 满了 → 扩容 return true; } // 出队 — 头部取出 public E poll() { E result = (E) elements[head]; if (result == null) return null; elements[head] = null; // 帮助 GC head = (head + 1) & (elements.length - 1); // 循环前进 return result; } 为什么用 (tail + 1) & (length - 1) 而不是 ？ - 数组长度始终是 2 的幂（如 8、16、32） - x & (length - 1) 等价于 x % length，但位运算快得多 扩容机制： java private void doubleCapacity() { int newCapacity = elements.length << 1; // 翻倍 Object[] newElements = new Object[newCapacity]; // 把 head..end 的部分搬到新数组前半段 System.arraycopy(elements, head, newElements, 0, elements.length - head); // 把 0..head 的部分搬到新数组后半段 System.arraycopy(elements, 0, newElements, elements.length - head, head); elements = newElements; head = 0; tail = oldCapacity; } --- ### 2. LinkedList — 双向链表队列 java public class LinkedList<E> { transient Node<E> first; // 头节点 transient Node<E> last; // 尾节点 transient int size = 0; private static class Node<E> { E item; Node<E> next; Node<E> prev; } } 入队（尾插）和出队（头删）： 入队 offer(E): null ← [A] ⇄ [B] ⇄ [C] → null ↓ linkLast(e): last.next = newNode newNode.prev = last last = newNode 出队 poll(): [A] ⇄ [B] ⇄ [C] ↓ unlinkFirst(): first = first.next first.prev = null return oldFirst.item java // 关键：都是 O(1) 操作 public boolean offer(E e) { linkLast(e); return true; } public E poll() { final Node<E> f = first; return (f == null) ? null : unlinkFirst(f); } 对比 ArrayDeque： | 维度 | ArrayDeque | LinkedList | |------|-----------|------------| | 底层结构 | 循环数组 | 双向链表 | | 内存 | 连续，缓存友好 | 分散，每个节点额外 24 字节（两个指针+对象头） | | 入队/出队 | 均摊 O(1)，偶尔扩容 O(n) | 始终 O(1) | | 适用场景 | 通用首选 | 频繁在中间插入删除时 | --- ### 3. PriorityQueue — 优先队列（小顶堆） 元素不是按入队顺序出队，而是按优先级（自然排序或 Comparator）出队。 底层：Object[] 数组，按**完全二叉堆**排列 1 ← 最小值在堆顶，最先出队 / \ 3 5 / \ / 7 4 8 数组表示: [null, 1, 3, 5, 7, 4, 8] 0 1 2 3 4 5 6 入队 — 上浮（siftUp）： java public boolean offer(E e) { int i = size; size = i + 1; if (i == 0) queue = e; else siftUp(i, e); // 放到末尾，然后上浮 return true; } private void siftUp(int k, E x) { while (k > 0) { int parent = (k - 1) >>> 1; // 父节点索引 E e = (E) queue[parent]; if (comparator.compare(x, e) >= 0) // x >= 父节点 → 到位了 break; queue[k] = e; // 父节点下移 k = parent; // 继续向上 } queue[k] = x; } 插入 2: 1 1 1 / \ → / \ → / \ 3 5 3 5 2 5 / \ / \ / \ / \ / \ / \ 7 4 8 2 7 4 8 7 4 8 ③ ↑ k=6 ↑ 与父(3)交换 出队 — 下沉（siftDown）： java public E poll() { E result = (E) queue; E x = (E) queue[--size]; queue[size] = null; if (size != 0) siftDown(0, x); // 把末尾元素放到堆顶，然后下沉 return result; } 弹出 1: ⑧ 3 3 / \ → / \ → / \ 2 5 2 5 2 5 / \ / / \ / \ 7 4 3 7 4 7 4 ↑ 末尾元素(8)放到堆顶 8>3 → 交换 8>4 → 停止（叶子节点） --- ### 4. 阻塞队列 — 生产者-消费者模型 #### ArrayBlockingQueue — 单锁 + Condition java public class ArrayBlockingQueue<E> { final Object[] items; // 循环数组 int takeIndex; // 出队位置 int putIndex; // 入队位置 int count; // 元素数量 final ReentrantLock lock; // 一把锁 private final Condition notEmpty; // 出队条件：队列非空 private final Condition notFull; // 入队条件：队列未满 } java // 入队 — 队满则阻塞等待 public void put(E e) throws InterruptedException { lock.lockInterruptibly(); try { while (count == items.length) notFull.await(); // 满了 → 阻塞生产者 enqueue(e); notEmpty.signal(); // 唤醒消费者 } finally { lock.unlock(); } } // 出队 — 队空则阻塞等待 public E take() throws InterruptedException { lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); // 空了 → 阻塞消费者 E e = dequeue(); notFull.signal(); // 唤醒生产者 return e; } finally { lock.unlock(); } } 生产者线程 共享缓冲区 消费者线程 ───────────────────────────────────────────────────────────────── put("A") ──lock──→ [A][ ][ ][ ] ──signal──→ take() → "A" put("B") ──lock──→ [A][B][ ][ ] ──signal──→ take() → "B" put("C") ──lock──→ [A][B][C][ ] ──signal──→ put("D") ──lock──→ [A][B][C][D] ──signal──→ take() → "A" put("E") ──await (满)──┐ take() → "B" ←──signal (有空位)──┘ #### LinkedBlockingQueue — 双锁提高吞吐 java public class LinkedBlockingQueue<E> { transient Node<E> head; // 哨兵头节点 private transient Node<E> last; private final AtomicInteger count = new AtomicInteger(0); // 两把锁分开控制！ final ReentrantLock takeLock = new ReentrantLock(); private final Condition notEmpty = takeLock.newCondition(); final ReentrantLock putLock = new ReentrantLock(); private final Condition notFull = putLock.newCondition(); private final int capacity; } 双锁的优势： 单锁 ArrayBlockingQueue: 生产者 ──┐ ┌── 消费者 ├── LOCK ── [队列] ──┤ ↑ 生产和消费互斥 ↑ 双锁 LinkedBlockingQueue: 生产者 ── putLock ──┐ ┌── takeLock ── 消费者 ↓ ↓ [ 队列 ] 生产者和消费者可以同时操作！ java // 入队 public void put(E e) throws InterruptedException { final ReentrantLock putLock = this.putLock; putLock.lockInterruptibly(); try { while (count.get() == capacity) notFull.await(); enqueue(e); int c = count.getAndIncrement(); if (c + 1 < capacity) // 还有空间 notFull.signal(); // 唤醒其他生产者 if (c == 0) // 之前是空的 signalNotEmpty(); // 唤醒消费者 } finally { putLock.unlock(); } } --- ### 5. ConcurrentLinkedQueue — 无锁 CAS 队列 基于 Michael-Scott 无锁算法，用 CAS 代替锁： java public class ConcurrentLinkedQueue<E> { private transient volatile Node<E> head; private transient volatile Node<E> tail; private static class Node<E> { volatile E item; volatile Node<E> next; } } java // 入队 — CAS 循环重试 public boolean offer(E e) { Node<E> newNode = new Node<>(e); for (;;) { // 无限重试 Node<E> t = tail; Node<E> p = t.next; // tail 可能滞后 if (p == null) { // tail 确实指向最后一个节点 if (t.casNext(null, newNode)) { // CAS 设置 next casTail(t, newNode); // 尝试推进 tail return true; } } else { // tail 滞后了，帮忙推进 casTail(t, p); } } } 多线程并发入队: 线程1: offer(X) 线程2: offer(Y) [A] → [B] → null [A] → [B] → null ↑t ↑t 线程1 CAS成功: 线程2 CAS失败，重试 → [A] → [B] → [X] [A] → [B] → [X] → null ↑t ↑t(线程1推进) 线程2 CAS 成功: [A] → [B] → [X] → [Y] ↑t(线程2推进) --- ## 四、全景对比 ┌─────────────────────┬──────────┬──────────┬────────┬──────────┐ │ 实现 │ 数据结构 │ 是否阻塞 │ 是否线程安全 │ 适用场景 │ ├─────────────────────┼──────────┼──────────┼────────┼──────────┤ │ ArrayDeque │ 循环数组 │ 否 │ 否 │ 单线程通用首选 │ │ LinkedList (as Q) │ 双向链表 │ 否 │ 否 │ 需要Deque接口 │ │ PriorityQueue │ 二叉堆 │ 否 │ 否 │ 优先级任务调度 │ │ ArrayBlockingQueue │ 循环数组 │ 是 │ 是(单锁)│ 有界生产消费 │ │ LinkedBlockingQueue │ 链表 │ 是 │ 是(双锁)│ 高吞吐生产消费 │ │ ConcurrentLinkedQ │ 链表 │ 否 │ 是(CAS)│ 无锁高并发 │ │ SynchronousQueue │ 无存储 │ 是 │ 是 │ 直接交接(handoff)│ │ PriorityBlockingQ │ 二叉堆 │ 是 │ 是 │ 带优先级的阻塞 │ │ DelayQueue │ 二叉堆 │ 是 │ 是 │ 定时/延迟任务 │ └─────────────────────┴──────────┴──────────┴────────┴──────────┘ ## 五、一句话总结 > ArrayDeque 用循环数组的头尾指针实现 O(1) 入出队；PriorityQueue 用二叉堆保证堆顶永远最优；阻塞队列用 Condition/锁 实现生产者-消费者的等待唤醒；ConcurrentLinkedQueue 用 CAS 自旋实现无锁并发 — 本质上都是在 数据结构 + 并发控制 两个维度上的不同组合。