阻塞队列 ⭐⭐

// BlockingQueue 接口中的定义
// 将指定元素插入此队列，必要时无限期等待可用空间
void put(E e) throws InterruptedException;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
public class ProducerConsumerDemo {
 
    public static void main(String[] args) {
        // 创建一个容量为 5 的有界阻塞队列
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
 
        // 生产者线程：不断地向队列中放入数据
        Thread producer = new Thread(() -> {
            try {
                int count = 0; // 消息计数器
                while (true) {
                    String message = "msg-" + count++; // 构造消息
                    System.out.println("[生产者] 准备放入: " + message);
                    queue.put(message); // 核心：如果队列满，线程在此阻塞
                    System.out.println("[生产者] 成功放入: " + message
                            + "，当前队列大小: " + queue.size());
                }
            } catch (InterruptedException e) {
                // put 被中断时会进入此分支
                Thread.currentThread().interrupt(); // 恢复中断标志
                System.out.println("[生产者] 被中断，优雅退出");
            }
        }, "Producer");
 
        // 消费者线程：不断地从队列中取出数据
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    Thread.sleep(1000); // 模拟消费慢于生产的场景
                    String message = queue.take(); // 核心：如果队列空，线程在此阻塞
                    System.out.println("  [消费者] 取出: " + message);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断标志
                System.out.println("  [消费者] 被中断，优雅退出");
            }
        }, "Consumer");
 
        producer.start(); // 启动生产者
        consumer.start(); // 启动消费者
    }
}

// BlockingQueue 接口中的定义
// 获取并移除此队列的头部，必要时无限期等待直到元素可用
E take() throws InterruptedException;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
public class MultiConsumerDemo {
 
    public static void main(String[] args) throws InterruptedException {
        // 容量为 3 的有界队列，用于传递任务
        BlockingQueue<String> taskQueue = new ArrayBlockingQueue<>(3);
 
        // 创建 3 个消费者线程，模拟工作线程池
        for (int i = 1; i <= 3; i++) {
            final int workerId = i; // effectively final, 供 lambda 捕获
            new Thread(() -> {
                try {
                    while (true) {
                        // 多个消费者同时调用 take()，竞争队头元素
                        // 底层的 ReentrantLock 保证同一时刻只有一个线程能拿到
                        String task = taskQueue.take();
                        System.out.println("Worker-" + workerId
                                + " 正在处理: " + task);
                        Thread.sleep(2000); // 模拟任务处理耗时
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // 恢复中断状态
                    System.out.println("Worker-" + workerId + " 退出");
                }
            }, "Worker-" + i).start();
        }
 
        // 主线程作为生产者，逐个提交 10 个任务
        for (int i = 1; i <= 10; i++) {
            String task = "Task-" + i; // 构造任务名
            taskQueue.put(task); // 如果队列满了就阻塞等待
            System.out.println("[主线程] 提交了: " + task);
        }
    }
}

// ArrayBlockingQueue.take() 的核心逻辑（简化版）
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock; // 获取实例锁引用
    lock.lockInterruptibly();             // 可中断地获取锁
    try {
        // 关键：使用 while 而非 if，防止虚假唤醒 (spurious wakeup)
        while (count == 0) {              // 如果队列为空
            notEmpty.await();             // 在 notEmpty 条件上等待
        }
        return dequeue();                 // 队列非空，执行出队操作
    } finally {
        lock.unlock();                    // 无论如何都要释放锁
    }
}

// 带超时的插入：等待指定时间，如果仍无法插入则返回 false
boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
 
// 带超时的获取：等待指定时间，如果仍无法获取则返回 null
E poll(long timeout, TimeUnit unit) throws InterruptedException;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
 
public class TimeoutDemo {
 
    public static void main(String[] args) {
        // 容量仅为 2 的队列，很容易满
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
 
        // === 演示 offer 超时 ===
        Thread producer = new Thread(() -> {
            try {
                // 前两个能立即成功
                System.out.println("offer A: " + queue.offer("A", 1, TimeUnit.SECONDS)); // true
                System.out.println("offer B: " + queue.offer("B", 1, TimeUnit.SECONDS)); // true
 
                // 第三个：队列已满，最多等 2 秒
                System.out.println("[生产者] 尝试 offer C，最多等 2 秒...");
                long start = System.currentTimeMillis();        // 记录开始时间
                boolean success = queue.offer("C", 2, TimeUnit.SECONDS); // 等2秒
                long elapsed = System.currentTimeMillis() - start; // 计算实际耗时
                System.out.println("[生产者] offer C 结果: " + success
                        + "，耗时: " + elapsed + "ms");
                // 如果 2 秒内没有消费者取走元素，success 为 false
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断标志
            }
        }, "Producer");
 
        // === 演示 poll 超时 ===
        Thread consumer = new Thread(() -> {
            try {
                // 等待 3 秒再开始消费，让生产者先感受超时
                Thread.sleep(3000);
 
                // 此时队列里有 A 和 B（C 大概率没放进去）
                System.out.println("  [消费者] poll: " + queue.poll(1, TimeUnit.SECONDS));
                System.out.println("  [消费者] poll: " + queue.poll(1, TimeUnit.SECONDS));
 
                // 队列已空，再 poll 会等 1 秒后返回 null
                System.out.println("  [消费者] poll 空队列（等 1 秒）: "
                        + queue.poll(1, TimeUnit.SECONDS)); // null
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断标志
            }
        }, "Consumer");
 
        producer.start(); // 启动生产者
        consumer.start(); // 启动消费者
    }
}

// ArrayBlockingQueue.offer(E, long, TimeUnit) 核心逻辑（简化版）
public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
    long nanos = unit.toNanos(timeout);       // 将超时时间统一转为纳秒
    final ReentrantLock lock = this.lock;     // 获取锁引用
    lock.lockInterruptibly();                 // 可中断地加锁
    try {
        while (count == items.length) {       // 队列满，进入循环等待
            if (nanos <= 0L) {                // 剩余等待时间用尽
                return false;                 // 返回 false 表示插入失败
            }
            // 在 notFull 条件上限时等待，返回值是剩余的纳秒数
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);                           // 队列未满（或被唤醒后仍有空位），入队
        return true;                          // 返回 true 表示插入成功
    } finally {
        lock.unlock();                        // 释放锁
    }
}

// 创建一个容量为 10 的阻塞队列
BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
 
    // 存储元素的定长数组，创建后长度不再改变
    final Object[] items;
 
    // 下一次 take/poll/peek 操作的数组下标
    int takeIndex;
 
    // 下一次 put/offer/add 操作的数组下标
    int putIndex;
 
    // 队列中当前元素的数量
    int count;
 
    // 唯一的那把锁
    final ReentrantLock lock;
 
    // 队列非空条件 —— 消费者在此等待
    private final Condition notEmpty;
 
    // 队列未满条件 —— 生产者在此等待
    private final Condition notFull;
}

// 环形数组示意（容量 = 6，当前有 3 个元素 A, B, C）
//
//        putIndex = 4
//            ↓
//   ┌───┬───┬───┬───┬───┬───┐
//   │   │ A │ B │ C │   │   │
//   └───┴───┴───┴───┴───┴───┘
//     0   1   2   3   4   5
//            ↑
//        takeIndex = 1
//
// 当 putIndex 到达 5 之后再入队，putIndex 会回绕到 0
// 逻辑上：  ... → [5] → [0] → [1] → ...  形成环

// 将下标向前推进一位，到达末尾时回绕到 0
final int inc(int i) {
    // 如果 i+1 等于数组长度，则返回 0；否则返回 i+1
    return (++i == items.length) ? 0 : i;
}

// 构造器 1：指定容量，默认非公平锁
public ArrayBlockingQueue(int capacity) {
    this(capacity, false); // 默认使用非公平模式
}
 
// 构造器 2：指定容量和公平性
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)                        // 容量必须为正整数
        throw new IllegalArgumentException();
    this.items = new Object[capacity];        // 分配定长数组
    lock = new ReentrantLock(fair);           // 根据 fair 参数创建锁
    notEmpty = lock.newCondition();           // 从同一把锁派生 notEmpty 条件
    notFull  = lock.newCondition();           // 从同一把锁派生 notFull 条件
}
 
// 构造器 3：用已有集合初始化队列
public ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);                     // 先调用构造器 2
    final ReentrantLock lock = this.lock;
    lock.lock();                              // 加锁保证可见性（不是为了互斥）
    try {
        final Object[] items = this.items;
        int i = 0;
        try {
            for (E e : c) {                   // 逐个将集合元素放入数组
                if (e == null)                // 不允许 null 元素
                    throw new NullPointerException();
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException(); // 集合大小超过容量
        }
        count = i;                            // 设置元素计数
        putIndex = (i == capacity) ? 0 : i;   // 设置下一次插入位置
    } finally {
        lock.unlock();
    }
}

public void put(E e) throws InterruptedException {
    Objects.requireNonNull(e);                // 不允许插入 null
    final ReentrantLock lock = this.lock;     // 引用局部变量（JIT 优化友好）
    lock.lockInterruptibly();                 // 可中断地获取锁
    try {
        while (count == items.length)         // 队列已满，进入等待循环
            notFull.await();                  // 在 notFull 条件上阻塞
        enqueue(e);                           // 真正执行入队
    } finally {
        lock.unlock();                        // 确保释放锁
    }
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;     // 同一把锁
    lock.lockInterruptibly();                 // 可中断地获取锁
    try {
        while (count == 0)                    // 队列为空，进入等待循环
            notEmpty.await();                 // 在 notEmpty 条件上阻塞
        return dequeue();                     // 真正执行出队
    } finally {
        lock.unlock();                        // 确保释放锁
    }
}

// 单锁瓶颈示意：
//
// 时间轴 ──────────────────────────────────────►
//
// Producer:  ████ lock ████ enqueue ████ unlock ████
//                                                    ████ lock ...
// Consumer:  ░░░░░░░░░░ BLOCKED ░░░░░░░░░░░░░░░░░░░░
//                                                    (等 Producer 释放)
//
// 理想状态（双锁，如 LinkedBlockingQueue）：
//
// Producer:  ████ putLock ████ enqueue ████ unlock ████
// Consumer:  ████ takeLock ████ dequeue ████ unlock ████
//            ↑ 两者可以并行！

// 两个 Condition 均从同一把 ReentrantLock 创建
final ReentrantLock lock = new ReentrantLock(fair);
final Condition notEmpty = lock.newCondition(); // 消费者等待的条件："队列非空"
final Condition notFull  = lock.newCondition(); // 生产者等待的条件："队列未满"

// 入队操作 —— 在持有锁的前提下调用
private void enqueue(E e) {
    final Object[] items = this.items;    // 取数组引用
    items[putIndex] = e;                  // 将元素放入 putIndex 位置
    if (++putIndex == items.length)       // putIndex 前进一步
        putIndex = 0;                     // 到达末尾，回绕到 0（环形）
    count++;                              // 元素数量 +1
    notEmpty.signal();                    // ★ 唤醒一个在 notEmpty 上等待的消费者
}
 
// 出队操作 —— 在持有锁的前提下调用
private E dequeue() {
    final Object[] items = this.items;    // 取数组引用
    @SuppressWarnings("unchecked")
    E e = (E) items[takeIndex];           // 取出 takeIndex 位置的元素
    items[takeIndex] = null;              // 将该位置置 null，帮助 GC
    if (++takeIndex == items.length)      // takeIndex 前进一步
        takeIndex = 0;                    // 到达末尾，回绕到 0（环形）
    count--;                              // 元素数量 -1
    // 如果有迭代器在工作，还需要通知迭代器
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();                     // ★ 唤醒一个在 notFull 上等待的生产者
    return e;                             // 返回取出的元素
}

// put 方法中
while (count == items.length)  // ★ 必须用 while，不能用 if
    notFull.await();
 
// take 方法中
while (count == 0)             // ★ 必须用 while，不能用 if
    notEmpty.await();

// 带超时的 offer：等待指定时间后若仍无法插入则返回 false
public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
    Objects.requireNonNull(e);                       // 不允许 null
    long nanos = unit.toNanos(timeout);              // 统一转换为纳秒
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();                        // 可中断获取锁
    try {
        while (count == items.length) {              // 队列满
            if (nanos <= 0L)                         // 超时时间耗尽
                return false;                        // 插入失败，返回 false
            nanos = notFull.awaitNanos(nanos);       // 等待指定纳秒数
            // awaitNanos 返回剩余时间（可能为负数）
        }
        enqueue(e);                                  // 成功入队
        return true;
    } finally {
        lock.unlock();
    }
}
 
// 带超时的 poll：等待指定时间后若仍无法取出则返回 null
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 <= 0L)                         // 超时时间耗尽
                return null;                         // 取出失败，返回 null
            nanos = notEmpty.awaitNanos(nanos);      // 等待指定纳秒数
        }
        return dequeue();                            // 成功出队
    } finally {
        lock.unlock();
    }
}

public class ProducerConsumerDemo {
 
    public static void main(String[] args) {
        // 创建容量为 5 的有界阻塞队列
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
 
        // 生产者线程：持续生产数据
        Thread producer = new Thread(() -> {
            int i = 0;                                   // 计数器
            try {
                while (true) {
                    String item = "item-" + (i++);       // 生成数据
                    queue.put(item);                      // 阻塞式插入（满则等待）
                    System.out.println("[Producer] put: " + item
                        + " | queue size: " + queue.size());
                    Thread.sleep(200);                    // 模拟生产耗时
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();       // 恢复中断标志
                System.out.println("[Producer] interrupted, exiting.");
            }
        }, "Producer");
 
        // 消费者线程：持续消费数据
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String item = queue.take();           // 阻塞式获取（空则等待）
                    System.out.println("[Consumer] take: " + item
                        + " | queue size: " + queue.size());
                    Thread.sleep(800);                    // 模拟消费耗时（慢于生产）
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();       // 恢复中断标志
                System.out.println("[Consumer] interrupted, exiting.");
            }
        }, "Consumer");
 
        producer.start();                                // 启动生产者
        consumer.start();                                // 启动消费者
    }
}

// LinkedBlockingQueue 的继承关系
public class LinkedBlockingQueue<E>
    extends AbstractQueue<E>           // 继承 AbstractQueue，复用部分模板方法
    implements BlockingQueue<E>,       // 实现 BlockingQueue 接口，提供阻塞语义
               java.io.Serializable   // 支持序列化

// LinkedBlockingQueue 的内部链表节点
static class Node<E> {
    E item;           // 存储的元素，为 null 时表示该节点已被逻辑删除或是哨兵节点
    Node<E> next;     // 指向下一个节点的引用（单向链表）
    Node(E x) {       // 构造函数，传入元素
        item = x;
    }
}

// 队列为空时的内存结构
┌─────────────────────────────────┐
│     LinkedBlockingQueue         │
│                                 │
│   head ──► [null|next:null]     │  ◄── 哨兵节点，item 永远为 null
│              ▲                  │
│   last ──────┘                  │  ◄── head 和 last 指向同一个哨兵
│                                 │
│   count = 0                     │
│   capacity = Integer.MAX_VALUE  │  ◄── 未指定容量时的默认值
└─────────────────────────────────┘

// 入队 3 个元素 A → B → C 后的内存结构
┌───────────────────────────────────────────────────────────────────┐
│     LinkedBlockingQueue                                           │
│                                                                   │
│   head ──► [null|next] ──► [A|next] ──► [B|next] ──► [C|null]   │
│             哨兵节点        第1个元素     第2个元素     第3个元素   │
│                                                        ▲          │
│   last ────────────────────────────────────────────────┘          │
│                                                                   │
│   count = 3                                                       │
└───────────────────────────────────────────────────────────────────┘

// take() 取出 A 之后的内存结构
┌───────────────────────────────────────────────────────┐
│     LinkedBlockingQueue                                │
│                                                        │
│   [旧哨兵] ──(GC回收)                                  │
│                                                        │
│   head ──► [null|next] ──► [B|next] ──► [C|null]      │
│            新哨兵(原A节点)   第1个元素     第2个元素     │
│                                            ▲           │
│   last ────────────────────────────────────┘           │
│                                                        │
│   count = 2                                            │
└───────────────────────────────────────────────────────┘

// 构造函数1：不指定容量 → 默认为 Integer.MAX_VALUE（约21亿）
// 实践中几乎等于"无界"，但潜藏 OOM 风险！
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);  // 默认容量设为 int 最大值
}
 
// 构造函数2：指定容量 → 有界队列
// 这是生产环境推荐的使用方式
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException(); // 容量必须为正数
    this.capacity = capacity;     // 设定最大容量
    last = head = new Node<E>(null);  // 初始化哨兵节点，head 和 last 同时指向它
}

// ⚠️ 反面教材：Executors.newFixedThreadPool 的实现
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(
        nThreads,                          // 核心线程数
        nThreads,                          // 最大线程数（等于核心线程数）
        0L, TimeUnit.MILLISECONDS,         // 无超时
        new LinkedBlockingQueue<Runnable>() // ⚠️ 无界队列！任务可以无限堆积
    );
}
 
// ✅ 正确做法：手动指定容量
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(1000); // 明确上界
ExecutorService executor = new ThreadPoolExecutor(
    4, 8, 60L, TimeUnit.SECONDS, workQueue  // 使用有界队列
);

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
 
    // ==================== 容量与计数 ====================
    private final int capacity;            // 队列最大容量（构造时设定，不可变）
    private final AtomicInteger count      // 当前元素数量（注意：使用 AtomicInteger 而非普通 int）
        = new AtomicInteger();
 
    // ==================== 链表头尾指针 ====================
    transient Node<E> head;                // 哨兵头节点（item 永远为 null）
    private transient Node<E> last;        // 尾节点（最后入队的元素）
 
    // ==================== 消费者端的锁 ====================
    private final ReentrantLock takeLock    // take 操作专用锁
        = new ReentrantLock();
    private final Condition notEmpty        // 条件变量：队列非空时唤醒等待的消费者
        = takeLock.newCondition();
 
    // ==================== 生产者端的锁 ====================
    private final ReentrantLock putLock     // put 操作专用锁
        = new ReentrantLock();
    private final Condition notFull         // 条件变量：队列未满时唤醒等待的生产者
        = putLock.newCondition();
}

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();  // 不允许插入 null 元素
    int c = -1;                       // 局部变量，记录操作前的元素数量，初始化为 -1 表示未赋值
    Node<E> node = new Node<E>(e);    // ① 在获取锁之前就创建好节点，减少持锁时间
    final ReentrantLock putLock = this.putLock;  // 引用到局部变量（JIT 优化惯例）
    final AtomicInteger count = this.count;       // 同上
 
    putLock.lockInterruptibly();       // ② 获取 putLock，可中断
    try {
        // ③ 使用 while 循环检查是否已满（防止虚假唤醒 spurious wakeup）
        while (count.get() == capacity) {
            notFull.await();           // 队列已满，当前生产者线程在 notFull 条件上阻塞等待
        }
        enqueue(node);                 // ④ 将新节点挂接到链表尾部
        c = count.getAndIncrement();   // ⑤ 原子递增 count，c 保存的是递增前的旧值
        if (c + 1 < capacity)          // ⑥ 如果入队后队列仍未满
            notFull.signal();          //    唤醒另一个可能在等待的生产者（级联唤醒）
    } finally {
        putLock.unlock();              // ⑦ 释放 putLock
    }
    if (c == 0)                        // ⑧ 如果入队前队列为空（c == 0 → 入队后变成了1个元素）
        signalNotEmpty();              //    需要唤醒可能正在等待的消费者
}

// 唤醒等待在 notEmpty 上的消费者
// 注意：此方法在 putLock 已释放后调用
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;  // 引用到局部变量
    takeLock.lock();          // 获取 takeLock（因为 notEmpty 是 takeLock 的条件变量）
    try {
        notEmpty.signal();    // 唤醒一个等待中的消费者
    } finally {
        takeLock.unlock();    // 释放 takeLock
    }
}

public E take() throws InterruptedException {
    E x;                               // 用于存放取出的元素
    int c = -1;                        // 局部变量，记录操作前的元素数量
    final AtomicInteger count = this.count;       // 引用到局部变量
    final ReentrantLock takeLock = this.takeLock;  // 引用到局部变量
 
    takeLock.lockInterruptibly();       // ① 获取 takeLock，可中断
    try {
        // ② 循环检查队列是否为空（防止虚假唤醒）
        while (count.get() == 0) {
            notEmpty.await();           // 队列为空，消费者在 notEmpty 条件上阻塞等待
        }
        x = dequeue();                  // ③ 从链表头部取出元素
        c = count.getAndDecrement();    // ④ 原子递减 count，c 保存递减前的旧值
        if (c > 1)                      // ⑤ 如果取出前队列中还有超过1个元素（取出后仍非空）
            notEmpty.signal();          //    级联唤醒另一个可能在等待的消费者
    } finally {
        takeLock.unlock();              // ⑥ 释放 takeLock
    }
    if (c == capacity)                  // ⑦ 如果取出前队列是满的（c == capacity → 取出后有1个空位）
        signalNotFull();               //    唤醒可能正在等待的生产者
    return x;                           // ⑧ 返回取出的元素
}

// 从链表头部取出元素（调用时已持有 takeLock）
private E dequeue() {
    Node<E> h = head;          // h 指向当前的哨兵节点
    Node<E> first = h.next;    // first 是真正的第一个数据节点
    h.next = h;                // 旧哨兵的 next 指向自身，帮助 GC（断开引用链）
    head = first;              // 将 first 提升为新的哨兵节点
    E x = first.item;         // 取出 first 中存储的元素
    first.item = null;         // 将新哨兵的 item 置为 null（符合哨兵约定）
    return x;                  // 返回取出的元素
}

public boolean remove(Object o) {
    if (o == null) return false;       // null 元素不存在于队列中
    fullyLock();                        // 同时获取 putLock 和 takeLock
    try {
        // 遍历链表，寻找目标元素
        for (Node<E> trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
            if (o.equals(p.item)) {     // 找到目标
                unlink(p, trail);       // 从链表中移除
                return true;
            }
        }
        return false;                   // 未找到
    } finally {
        fullyUnlock();                  // 释放两把锁
    }
}
 
// 同时获取两把锁（固定顺序，防止死锁）
void fullyLock() {
    putLock.lock();    // 先获取 putLock
    takeLock.lock();   // 再获取 takeLock
}
 
// 同时释放两把锁（与获取顺序相反）
void fullyUnlock() {
    takeLock.unlock(); // 先释放 takeLock
    putLock.unlock();  // 再释放 putLock
}

// ==================== 吞吐量优势的代价 ====================
 
// 代价1：更高的 GC 压力
// 每次入队都要 new Node()，每次出队都会产生需要回收的旧节点
// 在高吞吐场景下，短命对象的创建和回收会给 Young GC 带来显著压力
// ArrayBlockingQueue 则完全不产生额外对象
 
// 代价2：count 使用 AtomicInteger 带来的额外开销
// 虽然 CAS 操作很轻量，但在极高并发下，CPU Cache Line 的频繁失效
// (Cache Line Bouncing) 仍有一定影响
 
// 代价3：更差的缓存局部性 (Cache Locality)
// 数组元素在内存中连续存放，CPU 可以利用预取 (Prefetch) 机制
// 链表节点散布在堆内存各处，每次访问 next 指针都可能触发 Cache Miss
 
// 代价4：每个节点的额外内存开销
// Node 对象本身的对象头（12字节，压缩指针下）+ item 引用 + next 引用
// 相比数组中一个引用槽位（4~8字节），单个元素的内存开销多出约 3~4 倍

// CachedThreadPool 的源码定义，其核心就是 SynchronousQueue
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(
        0,                          // corePoolSize: 核心线程为 0
        Integer.MAX_VALUE,          // maximumPoolSize: 最大线程数几乎无限
        60L, TimeUnit.SECONDS,      // keepAliveTime: 空闲线程存活 60 秒
        new SynchronousQueue<Runnable>()  // workQueue: 零容量同步队列
    );
}

// 基本特征验证
SynchronousQueue<String> sq = new SynchronousQueue<>();
 
// size() 永远返回 0
System.out.println(sq.size());        // 输出: 0
 
// isEmpty() 永远返回 true
System.out.println(sq.isEmpty());     // 输出: true
 
// peek() 永远返回 null（没有元素可窥探）
System.out.println(sq.peek());        // 输出: null
 
// remainingCapacity() 永远返回 0
System.out.println(sq.remainingCapacity()); // 输出: 0
 
// offer() 在无匹配线程时立即返回 false（非阻塞）
System.out.println(sq.offer("hello")); // 输出: false
 
// poll() 在无匹配线程时立即返回 null（非阻塞）
System.out.println(sq.poll());         // 输出: null
 
// iterator() 返回空迭代器
System.out.println(sq.iterator().hasNext()); // 输出: false

public class DirectHandoffDemo {
    public static void main(String[] args) {
        // 创建同步队列
        SynchronousQueue<String> channel = new SynchronousQueue<>();
 
        // 消费者线程：先启动，阻塞等待生产者
        Thread consumer = new Thread(() -> {
            try {
                System.out.println("[Consumer] 等待数据...");
                long start = System.currentTimeMillis();       // 记录开始等待的时间
 
                String data = channel.take();                  // 阻塞，直到生产者 put
 
                long elapsed = System.currentTimeMillis() - start;
                System.out.println("[Consumer] 收到: " + data  // 打印接收到的数据
                    + "，等待了 " + elapsed + "ms");            // 打印等待耗时
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();             // 恢复中断标志
            }
        }, "consumer-thread");
 
        // 生产者线程：延迟 2 秒后 put
        Thread producer = new Thread(() -> {
            try {
                Thread.sleep(2000);                            // 模拟延迟 2 秒
                System.out.println("[Producer] 发送数据...");
                channel.put("Hello SynchronousQueue!");        // 阻塞，直到消费者 take
                System.out.println("[Producer] 数据已被消费者取走"); // put 返回意味着握手成功
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "producer-thread");
 
        consumer.start();                                      // 先启动消费者
        producer.start();                                      // 再启动生产者
    }
}
 
// 输出:
// [Consumer] 等待数据...
// (... 约 2 秒后 ...)
// [Producer] 发送数据...
// [Consumer] 收到: Hello SynchronousQueue!，等待了 2003ms
// [Producer] 数据已被消费者取走

// 非公平模式（默认）：内部使用 TransferStack（栈 / LIFO）
SynchronousQueue<String> unfairQueue = new SynchronousQueue<>();
SynchronousQueue<String> unfairQueue2 = new SynchronousQueue<>(false);
 
// 公平模式：内部使用 TransferQueue（队列 / FIFO）
SynchronousQueue<String> fairQueue = new SynchronousQueue<>(true);

public SynchronousQueue(boolean fair) {
    // fair=true  → 使用 TransferQueue（FIFO，先来先服务）
    // fair=false → 使用 TransferStack（LIFO，后来先服务）
    transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}

// TransferStack.transfer() 的简化伪代码
E transfer(E e, boolean timed, long nanos) {
    SNode s = null;                                // 当前线程将要创建的节点
    // e==null 表示 take（REQUEST），e!=null 表示 put（DATA）
    int mode = (e == null) ? REQUEST : DATA;
 
    for (;;) {                                     // 经典 CAS 自旋循环
        SNode h = head;                            // 读取栈顶节点
 
        if (h == null || h.mode == mode) {
            // 情况1: 栈为空 或 栈顶节点与当前同类（都是生产者或都是消费者）
            // → 将当前节点压入栈顶，然后阻塞等待配对
            // 注意: 压到栈顶 = LIFO = 后来的反而先被匹配 = 非公平
            s = new SNode(mode, e);                // 创建新节点
            if (casHead(h, s)) {                   // CAS 更新栈顶
                // 自旋一段时间后 park 阻塞，等待被匹配
                SNode m = awaitFulfill(s, timed, nanos);
                return (mode == REQUEST) ? m.item : e;
            }
        }
        else if (!isFulfilling(h.mode)) {
            // 情况2: 栈顶节点是互补类型（一个 REQUEST 一个 DATA）且未在匹配中
            // → 尝试匹配：创建一个 FULFILLING 节点压入栈顶
            s = new SNode(FULFILLING | mode, e);   // 设置 FULFILLING 标记
            if (casHead(h, s)) {                   // CAS 更新栈顶
                SNode m = h;                       // m 是被匹配的节点
                // 完成匹配: 交换数据，唤醒对方线程，弹出两个节点
                casHead(s, m.next);                // 弹出 s 和 m
                // 返回交换到的数据
                return (mode == REQUEST) ? m.item : e;
            }
        }
        else {
            // 情况3: 栈顶正在 FULFILLING，帮助它完成匹配
            // （无锁编程中的"帮助完成"模式）
            // ...
        }
    }
}

// TransferQueue.transfer() 的简化伪代码
E transfer(E e, boolean timed, long nanos) {
    QNode s = null;
    boolean isData = (e != null);                  // 判断是生产者还是消费者
 
    for (;;) {
        QNode t = tail;                            // 读取队尾
        QNode h = head;                            // 读取队头
 
        if (t == h || t.isData == isData) {
            // 情况1: 队列为空 或 队尾节点与当前同类
            // → 将当前节点加入队尾，然后阻塞等待配对
            // 加入队尾 = FIFO = 先来的先被匹配 = 公平
            s = new QNode(e, isData);              // 创建新节点
            if (t.casNext(null, s)) {              // CAS 追加到队尾
                casTail(t, s);                     // CAS 更新 tail 指针
                // 自旋后 park 阻塞
                Object x = awaitFulfill(s, e, timed, nanos);
                return (E) x;
            }
        }
        else {
            // 情况2: 队头节点是互补类型
            // → 从队头开始匹配（FIFO 保证公平性）
            QNode m = h.next;                      // 取队头后第一个真实节点
            if (m != null && m.isData != isData) {
                // 尝试匹配: CAS 交换数据
                Object x = m.item;
                if (m.casItem(x, e)) {             // CAS 写入数据完成握手
                    advanceHead(h, m);             // 推进 head，弹出已匹配节点
                    LockSupport.unpark(m.waiter);  // 唤醒等待的对方线程
                    return (x != null) ? (E) x : e;
                }
            }
        }
    }
}

import java.util.concurrent.SynchronousQueue;
 
public class FairnessDemo {
    public static void main(String[] args) throws InterruptedException {
        // ============ 测试非公平模式 ============
        System.out.println("===== 非公平模式 (LIFO) =====");
        testFairness(false);                              // 传入 false 使用 TransferStack
 
        Thread.sleep(3000);                               // 等待上一轮测试完成
 
        // ============ 测试公平模式 ============
        System.out.println("\n===== 公平模式 (FIFO) =====");
        testFairness(true);                               // 传入 true 使用 TransferQueue
    }
 
    static void testFairness(boolean fair) throws InterruptedException {
        SynchronousQueue<String> queue = new SynchronousQueue<>(fair);
 
        // 依次启动 3 个生产者，每个间隔 100ms 确保入栈/队顺序可控
        for (int i = 1; i <= 3; i++) {
            final int id = i;                             // 捕获当前编号
            new Thread(() -> {
                try {
                    System.out.println("Producer-" + id + " 准备 put...");
                    queue.put("data-" + id);              // 阻塞，等待消费者匹配
                    System.out.println("Producer-" + id + " put 完成");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "Producer-" + id).start();
 
            Thread.sleep(100);                            // 确保按 1→2→3 的顺序进入等待
        }
 
        Thread.sleep(500);                                // 让 3 个生产者都进入等待状态
 
        // 启动 3 个消费者，依次 take
        for (int i = 1; i <= 3; i++) {
            final int id = i;
            new Thread(() -> {
                try {
                    String data = queue.take();           // 消费者取数据
                    System.out.println("Consumer-" + id
                        + " 取到: " + data);              // 观察取到的数据来自哪个生产者
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "Consumer-" + id).start();
 
            Thread.sleep(100);                            // 让匹配逐个发生，便于观察顺序
        }
    }
}
 
// ===== 非公平模式 (LIFO) 典型输出 =====
// Producer-1 准备 put...
// Producer-2 准备 put...
// Producer-3 准备 put...
// Consumer-1 取到: data-3    ← 后进的 Producer-3 反而先被匹配（栈顶）
// Consumer-2 取到: data-2
// Consumer-3 取到: data-1    ← 最先进的 Producer-1 最后被匹配（栈底）
 
// ===== 公平模式 (FIFO) 典型输出 =====
// Producer-1 准备 put...
// Producer-2 准备 put...
// Producer-3 准备 put...
// Consumer-1 取到: data-1    ← 先进的 Producer-1 先被匹配（队头）
// Consumer-2 取到: data-2
// Consumer-3 取到: data-3    ← 最后进的 Producer-3 最后被匹配（队尾）

import java.util.concurrent.PriorityBlockingQueue;
 
/**
 * 患者类，实现 Comparable 接口以支持优先级排序
 * 优先级数字越小，病情越紧急，应优先处理
 */
class Patient implements Comparable<Patient> {
    // 患者姓名
    private final String name;
    // 优先级：1=危重, 2=急症, 3=普通
    private final int priority;
    // 到达时间戳，同优先级时按到达先后排序
    private final long arrivalTime;
 
    public Patient(String name, int priority) {
        this.name = name;
        this.priority = priority;
        // 记录创建（到达）时间
        this.arrivalTime = System.nanoTime();
    }
 
    /**
     * 比较规则：
     * 1. 首先按 priority 升序（数字越小越紧急）
     * 2. 同 priority 时，按 arrivalTime 升序（先到先处理）
     */
    @Override
    public int compareTo(Patient other) {
        // 先比较优先级
        int result = Integer.compare(this.priority, other.priority);
        if (result == 0) {
            // 优先级相同时，比较到达时间
            result = Long.compare(this.arrivalTime, other.arrivalTime);
        }
        return result;
    }
 
    @Override
    public String toString() {
        // 将优先级数字映射为可读文本
        String level = switch (priority) {
            case 1 -> "危重";
            case 2 -> "急症";
            case 3 -> "普通";
            default -> "未知";
        };
        return String.format("[%s-%s] %s", level, priority, name);
    }
}
 
public class EmergencyRoomDemo {
    public static void main(String[] args) throws InterruptedException {
        // 创建优先级阻塞队列，无需指定容量
        PriorityBlockingQueue<Patient> queue = new PriorityBlockingQueue<>();
 
        // 模拟患者以任意顺序到达
        queue.put(new Patient("张三", 3));  // 普通患者先到
        queue.put(new Patient("李四", 1));  // 危重患者后到
        queue.put(new Patient("王五", 2));  // 急症患者
        queue.put(new Patient("赵六", 1));  // 另一个危重患者
        queue.put(new Patient("钱七", 3));  // 另一个普通患者
 
        // 按优先级出队
        System.out.println("=== 急诊分诊顺序 ===");
        while (!queue.isEmpty()) {
            // take() 会取出优先级最高（compareTo 最小）的元素
            Patient patient = queue.take();
            System.out.println(patient);
        }
    }
}

// 1. 默认构造：初始容量 11，自然排序
PriorityBlockingQueue<Task> q1 = new PriorityBlockingQueue<>();
 
// 2. 指定初始容量（注意：这只是"初始"容量，队列仍然无界）
PriorityBlockingQueue<Task> q2 = new PriorityBlockingQueue<>(100);
 
// 3. 指定初始容量 + 自定义比较器
// 这里用 Comparator 实现"数字大的优先级高"（最大堆）
PriorityBlockingQueue<Integer> q3 = new PriorityBlockingQueue<>(
    16,                              // 初始容量
    (a, b) -> Integer.compare(b, a)  // 降序比较器，形成最大堆
);
 
// 4. 从已有集合初始化
List<Task> existingTasks = List.of(task1, task2, task3);
PriorityBlockingQueue<Task> q4 = new PriorityBlockingQueue<>(existingTasks);

PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
queue.addAll(List.of(5, 1, 3, 2, 4));
 
// ❌ 迭代器不保证顺序！可能输出：1 2 3 5 4 或其他堆序排列
System.out.println("Iterator 遍历：");
for (int num : queue) {
    System.out.print(num + " ");
}
 
// ✅ 逐个 poll 才能保证优先级顺序：1 2 3 4 5
System.out.println("\npoll 逐个取出：");
while (!queue.isEmpty()) {
    System.out.print(queue.poll() + " ");
}

public class PriorityBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
 
    // 默认初始容量
    private static final int DEFAULT_INITIAL_CAPACITY = 11;
 
    // 数组最大容量（减 8 是因为某些 VM 会在数组头部保留几个字节）
    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
 
    // 存放元素的数组（堆的物理存储）
    // 堆顶元素始终在 queue[0]
    private transient Object[] queue;
 
    // 当前队列中的元素数量
    private transient int size;
 
    // 比较器，如果为 null 则使用元素的自然排序
    private transient Comparator<? super E> comparator;
 
    // 唯一的一把锁，保护所有公共操作
    private final ReentrantLock lock;
 
    // 条件变量：队列非空时通知等待的消费者
    // 注意：没有 notFull 条件！因为队列无界，永远不满
    private final Condition notEmpty;
 
    // 用于 CAS 控制扩容的自旋锁（0=未锁定, 1=已锁定）
    private transient volatile int allocationSpinLock;
}

// put 直接委托给 offer，永不阻塞
public void put(E e) {
    offer(e); // 永远不需要等待
}
 
public boolean offer(E e) {
    // 1. 空值检查——PriorityBlockingQueue 不允许 null 元素
    if (e == null)
        throw new NullPointerException();
 
    // 2. 获取独占锁
    final ReentrantLock lock = this.lock;
    lock.lock();
 
    int n, cap;
    Object[] es;
 
    // 3. 如果当前元素数量 >= 数组容量，需要扩容
    while ((n = size) >= (cap = (es = queue).length))
        tryGrow(es, cap);  // 扩容操作（详见下文）
 
    try {
        // 4. 获取比较器引用
        final Comparator<? super E> cmp;
 
        // 5. 执行 siftUp，将新元素放到正确位置
        if ((cmp = comparator) == null)
            siftUpComparable(n, e, es);      // 使用自然排序
        else
            siftUpUsingComparator(n, e, es, cmp); // 使用自定义比较器
 
        // 6. 元素数量 +1
        size = n + 1;
 
        // 7. 唤醒一个正在 take() 中等待的消费者线程
        notEmpty.signal();
    } finally {
        // 8. 释放锁
        lock.unlock();
    }
    return true; // 无界队列，插入永远成功
}

/**
 * 上浮操作：将新元素从位置 k 开始向上调整
 * @param k 新元素的初始位置（数组末尾）
 * @param x 要插入的新元素
 * @param es 堆数组
 */
private static <T> void siftUpComparable(int k, T x, Object[] es) {
    // 将元素转为 Comparable 类型
    Comparable<? super T> key = (Comparable<? super T>) x;
 
    // 持续上浮，直到到达堆顶（k == 0）
    while (k > 0) {
        // 计算父节点的下标：parent = (k - 1) / 2
        int parent = (k - 1) >>> 1;
 
        // 获取父节点元素
        Object e = es[parent];
 
        // 如果新元素 >= 父节点，堆序已满足，停止上浮
        if (key.compareTo((T) e) >= 0)
            break;
 
        // 否则，将父节点下移到位置 k
        es[k] = e;
 
        // 新元素继续往上走，k 移到父节点位置
        k = parent;
    }
    // 最终将新元素放到正确位置
    es[k] = key;
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    // 获取可中断的锁
    lock.lockInterruptibly();
    E result;
    try {
        // 核心：当 dequeue 返回 null（队列为空），持续等待
        while ((result = dequeue()) == null)
            notEmpty.await();  // 阻塞，等待 offer() 的 signal 唤醒
    } finally {
        lock.unlock();
    }
    return result;
}
 
/**
 * 出队操作：取出堆顶元素，并通过 siftDown 恢复堆序
 * 调用者必须已持有锁
 */
private E dequeue() {
    // n 是最后一个元素的下标
    final int n;
    // 如果队列为空，返回 null
    if ((n = size - 1) < 0)
        return null;
 
    final Object[] es = queue;
 
    // 1. 取出堆顶元素（优先级最高的）
    E result = (E) es[0];
 
    // 2. 取出数组最后一个元素，准备用它替换堆顶
    E x = (E) es[n];
 
    // 3. 将最后位置置空（help GC）
    es[n] = null;
 
    // 4. 更新 size
    size = n;
 
    // 5. 如果队列还有元素，执行 siftDown 恢复堆序
    if (n > 0) {
        final Comparator<? super E> cmp;
        if ((cmp = comparator) == null)
            siftDownComparable(0, x, es, n);      // 自然排序下沉
        else
            siftDownUsingComparator(0, x, es, n, cmp); // 比较器下沉
    }
    return result;
}

/**
 * 下沉操作：将元素从位置 k 开始向下调整
 * @param k   起始位置（通常为 0，即堆顶）
 * @param x   要下沉的元素
 * @param es  堆数组
 * @param n   堆的大小（有效元素个数）
 */
private static <T> void siftDownComparable(int k, T x, Object[] es, int n) {
    Comparable<? super T> key = (Comparable<? super T>) x;
 
    // half = n/2，即第一个叶子节点的下标
    // 只有非叶子节点才需要下沉（叶子节点无子节点可比较）
    int half = n >>> 1;
 
    while (k < half) {
        // 左子节点下标
        int child = (k << 1) + 1;   // 等价于 2*k + 1
 
        // 假设左子节点更小
        Object c = es[child];
 
        // 右子节点下标
        int right = child + 1;
 
        // 如果右子节点存在，且右子节点 < 左子节点
        // 则选择更小的那个子节点
        if (right < n &&
                ((Comparable<? super T>) c).compareTo((T) es[right]) > 0)
            c = es[child = right];  // 更新 child 为右子节点
 
        // 如果当前元素 <= 较小的子节点，堆序已满足，停止下沉
        if (key.compareTo((T) c) <= 0)
            break;
 
        // 否则，将较小的子节点上移到位置 k
        es[k] = c;
 
        // 继续下沉到子节点的位置
        k = child;
    }
    // 将元素放到最终位置
    es[k] = key;
}