核心概念：阻塞 vs. 非阻塞

在选择线程安全的 Queue 之前，首先要理解一个关键区别：阻塞 和 非阻塞。

• 阻塞队列:

  • 当队列为空时，从队列中获取元素（take() 或 poll(long timeout, TimeUnit unit)）的线程将会被阻塞,直到有新的元素被放入队列。
  • 当队列已满时，向队列中添加元素（put() 或 offer(long timeout, TimeUnit unit)）的线程将会被阻塞,直到队列中有空闲位置。
  • 适用场景：生产者-消费者模型，这是最常见的使用场景，可以优雅地协调线程间的任务传递，避免忙等待,提高效率。

• 非阻塞队列:

  • 当队列为空时，尝试获取元素的线程会立即返回 null 或抛出异常,而不会被阻塞。
  • 当队列已满时，尝试添加元素的线程会立即返回 false 或抛出异常,而不会被阻塞。
  • 适用场景：需要快速响应、不希望线程因等待而挂起的场景，通常与 java.util.concurrent.atomic 包下的原子类结合使用,实现无锁算法。

主要的线程安全 Queue 实现

Java 的 java.util.concurrent 包提供了丰富的线程安全 Queue 实现。

A. 阻塞队列

LinkedBlockingQueue

• 实现: 基于链表的有界阻塞队列。
• 特点:
  • 容量可配置: 在创建时可以指定容量，如果不指定，则默认为 Integer.MAX_VALUE,相当于无界队列。
  • 高吞吐量: 在生产者-消费者场景下,性能通常非常好。
  • 双锁分离: 内部使用两个独立的 ReentrantLock 分别控制入队和出队，减少了锁竞争,提高了并发性能。
• 适用场景: 最通用的生产者-消费者队列,适用于大多数异步任务处理场景。

ArrayBlockingQueue

• 实现: 基于数组的有界阻塞队列。
• 特点:
  • 固定容量: 必须在创建时指定容量。
  • FIFO: 严格按照先进先出的原则排序。
  • 单锁: 内部只使用一个 ReentrantLock，同时控制入队和出队，这意味着即使一个线程在入队，另一个线程也无法出队,反之亦然。
• 适用场景: 当你需要一个固定大小的队列，并且希望它能够限制任务的积压时，性能通常略低于 LinkedBlockingQueue,因为锁粒度更粗。

PriorityBlockingQueue

• 实现: 无界的优先级阻塞队列。
• 特点:
  • 无界: 理论上可以无限增长,因此要小心内存溢出。
  • 优先级: 元素按照自然排序或提供的 Comparator 进行排序。poll() 或 take() 总是获取队列中优先级最高的元素。
  • 基于堆: 内部使用堆数据结构来维护优先级。
• 适用场景: 需要按照特定优先级处理任务的场景，例如操作系统的任务调度、网络数据包的优先级发送等。

SynchronousQueue

• 实现: 一个特殊的阻塞队列，它不存储任何元素。
• 特点:
  • 一对一交接: 每个插入操作必须等待另一个线程的对应的移除操作，反之亦然，当一个线程尝试向 SynchronousQueue 放入一个元素时,它必须等待另一个线程来取走这个元素。
  • 无缓冲: 它更像一个“会合点”或“rendezvous point”,而不是一个队列。
• 适用场景: 适用于高并发下的“直接传递”（handoff）场景，性能极高。ThreadPoolExecutor 中的 newCachedThreadPool 就使用了它来创建线程，当有新任务提交时，会尝试将任务直接交给空闲线程，如果没有空闲线程,则创建新线程。

DelayQueue

• 实现: 无界阻塞队列，其元素必须实现 java.util.concurrent.Delayed 接口。
• 特点:
  • 延迟消费: 只有当元素的延迟时间到期后,才能从队列中获取到它。
  • 按延迟排序: 队列中延迟时间最短（即将到期）的元素会最先被取出。
• 适用场景: 实现定时任务、缓存过期清理、订单超时自动取消等场景。

B. 非阻塞队列

ConcurrentLinkedQueue

• 实现: 基于链表的非阻塞并发队列。
• 特点:
  • 无界: 可以无限增长。
  • 高并发: 使用 CAS (Compare-And-Swap) 操作实现，完全避免了锁,因此在极高并发下性能非常出色。
  • 弱一致性: 在遍历队列时，可能无法反映出队列的最新状态,因为遍历过程不会阻塞其他操作。
• 适用场景: 在高并发环境下，需要一个高性能的 FIFO 队列，且不关心遍历时的一致性，用于缓存、日志收集等。

ConcurrentLinkedDeque

• 实现: 基于双向链表的非阻塞并发双端队列。
• 特点:
  • 功能与 ConcurrentLinkedQueue 类似，但支持在两端进行操作（addFirst, addLast, pollFirst, pollLast）。
  • 同样使用 CAS 操作，无锁,性能高。
• 适用场景: 需要高性能的双端队列的场景，例如工作窃取算法（ForkJoinPool 就用它来管理任务）。

性能与选择对比

如何选择？

1. 如果你在使用生产者-消费者模式:

   • 首选 LinkedBlockingQueue: 它是平衡了功能和性能的最佳选择。
   • 如果需要严格限制队列大小以防止内存溢出，使用 ArrayBlockingQueue。
   • 如果任务有优先级，使用 PriorityBlockingQueue。
   • 如果你希望即时传递，减少队列开销，使用 SynchronousQueue。

2. 如果你需要高并发且不希望线程阻塞:

   • 首选 ConcurrentLinkedQueue: 对于简单的 FIFO 队列,它的性能极高。
   • 如果你需要在队列两端进行操作，使用 ConcurrentLinkedDeque。

3. 如果你需要处理定时任务:

   • 使用 DelayQueue。

最佳实践与代码示例

示例 1: 生产者-消费者模型 (使用 LinkedBlockingQueue)

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
class Producer implements Runnable {
    private final BlockingQueue<String> queue;
    public Producer(BlockingQueue<String> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        try {
            for (int i = 0; i < 10; i++) {
                String task = "Task-" + i;
                System.out.println("Producer produced: " + task);
                queue.put(task); // 如果队列满，这里会阻塞
                TimeUnit.SECONDS.sleep(1); // 模拟生产耗时
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
class Consumer implements Runnable {
    private final BlockingQueue<String> queue;
    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        try {
            while (true) { // 持续消费
                String task = queue.take(); // 如果队列空，这里会阻塞
                System.out.println("Consumer consumed: " + task);
                // 模拟消费耗时
                TimeUnit.MILLISECONDS.sleep(500);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
public class BlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(5); // 设置容量为5
        Thread producerThread = new Thread(new Producer(queue));
        Thread consumerThread = new Thread(new Consumer(queue));
        producerThread.start();
        consumerThread.start();
        try {
            producerThread.join();
            // 为了优雅停止消费者，可以在这里设置一个标志位，或者使用外部机制
            // consumerThread.interrupt(); // 在实际应用中，可能需要一个更优雅的停止机制
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

示例 2: 高性能非阻塞队列 (使用 ConcurrentLinkedQueue)

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class ConcurrentLinkedQueueExample {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
        ExecutorService executor = Executors.newFixedThreadPool(10);
        // 10个生产者
        for (int i = 0; i < 10; i++) {
            final int id = i;
            executor.submit(() -> {
                for (int j = 0; j < 1000; j++) {
                    queue.add("Producer-" + id + "-Item-" + j);
                }
            });
        }
        // 5个消费者
        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    String item = queue.poll(); // 非阻塞，如果为空则返回null
                    if (item != null) {
                        // 处理 item...
                        // System.out.println("Processed: " + item);
                    } else {
                        // 如果队列为空，可以短暂休眠以避免CPU空转
                        try {
                            TimeUnit.MILLISECONDS.sleep(10);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    }
                }
            });
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        System.out.println("Final queue size: " + queue.size());
    }
}

• java.util.concurrent 包是线程安全编程的核心，其提供的 Queue 实现经过了精心设计和优化。
• 明确你的需求：是需要阻塞还是非阻塞？有界还是无界？FIFO 还是优先级？
• 生产者-消费者首选 LinkedBlockingQueue,它在大多数情况下提供了最佳的性能和功能平衡。
• 极高并发且无阻塞需求时，选择 ConcurrentLinkedQueue,它的无锁设计能带来极致的并发性能。
• 永远警惕无界队列，如 LinkedBlockingQueue（未指定容量时）、PriorityBlockingQueue 和 DelayQueue，它们可能导致内存溢出，在有界场景下,应优先选择有界队列。