Java中的并发容器：BlockingQueue详解

一、BlockingQueue概述
BlockingQueue是Java并发包(java.util.concurrent)中提供的一个线程安全的队列接口，它不仅具备普通队列的FIFO特性，更重要的是支持阻塞操作：当队列已满时，插入操作会被阻塞直到有空间可用；当队列为空时，获取操作会被阻塞直到有元素可用。

二、核心特性与使用场景

1. 线程安全：所有实现类都是线程安全的，无需额外同步
2. 生产者-消费者模式：完美解决生产者和消费者之间的协调问题
3. 流量控制：通过固定容量实现流量控制，防止内存溢出
4. 阻塞策略：提供多种阻塞、超时和非阻塞的操作方法

三、主要方法对比

| 操作类型 | 抛出异常  | 返回特殊值 | 阻塞     | 超时               |
|----------|-----------|------------|----------|--------------------|
| 插入     | add(e)    | offer(e)   | put(e)   | offer(e,time,unit) |
| 移除     | remove()  | poll()     | take()   | poll(time,unit)    |
| 检查     | element() | peek()     | 不可用   | 不可用             |

四、常用实现类详解

1. ArrayBlockingQueue

• 底层结构：基于数组的有界阻塞队列
• 特点：
  • 必须指定容量，一旦创建大小不可变
  • 默认使用非公平锁，可选的公平锁策略
  • 内部使用ReentrantLock和Condition实现线程安全

// 创建示例
BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

// 生产者线程
queue.put("item"); // 阻塞插入
boolean success = queue.offer("item", 1, TimeUnit.SECONDS); // 超时插入

// 消费者线程
String item = queue.take(); // 阻塞获取
String item = queue.poll(1, TimeUnit.SECONDS); // 超时获取

2. LinkedBlockingQueue

• 底层结构：基于链表的可选有界队列
• 特点：
  • 如果不指定容量，默认为Integer.MAX_VALUE（近似无界）
  • 采用两把锁（putLock和takeLock）分离读写操作，提高并发性能
  • 适合生产消费速度差异较大的场景

// 无界队列
BlockingQueue<Integer> unboundedQueue = new LinkedBlockingQueue<>();

// 有界队列（容量100）
BlockingQueue<Integer> boundedQueue = new LinkedBlockingQueue<>(100);

3. PriorityBlockingQueue

• 底层结构：基于堆的无界优先级队列
• 特点：
  • 无界队列，但会自动扩容
  • 元素必须实现Comparable接口或提供Comparator
  • 保证出队顺序按照优先级，而不是FIFO

// 创建优先级队列
BlockingQueue<Task> priorityQueue = new PriorityBlockingQueue<>();

// 自定义比较器
BlockingQueue<Task> customPriorityQueue = 
    new PriorityBlockingQueue<>(11, Comparator.comparing(Task::getPriority));

4. SynchronousQueue

• 底层结构：不存储元素的特殊队列
• 特点：
  • 每个插入操作必须等待对应的移除操作
  • 适合直接传递场景，吞吐量高于ArrayBlockingQueue和LinkedBlockingQueue
  • 可选择公平模式（队列）或非公平模式（栈）

BlockingQueue<String> syncQueue = new SynchronousQueue<>();

// 生产者线程
new Thread(() -> {
    try {
        syncQueue.put("data"); // 等待消费者获取
        System.out.println("数据已传递");
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

// 消费者线程  
new Thread(() -> {
    try {
        String data = syncQueue.take(); // 获取生产者放入的数据
        System.out.println("收到数据: " + data);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

五、实现原理深度解析

以ArrayBlockingQueue为例的源码分析：

public class ArrayBlockingQueue<E> {
    final Object[] items;        // 存储元素的数组
    int takeIndex;              // 下一个要获取的元素索引
    int putIndex;               // 下一个要放置的元素索引
    int count;                  // 队列中元素个数
    
    final ReentrantLock lock;   // 主锁
    private final Condition notEmpty;  // 等待获取的条件
    private final Condition notFull;   // 等待插入的条件
    
    // put方法实现
    public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();  // 获取可中断锁
        try {
            while (count == items.length)  // 队列已满
                notFull.await();           // 在notFull条件上等待
            enqueue(e);                    // 入队操作
        } finally {
            lock.unlock();
        }
    }
    
    // take方法实现  
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)     // 队列为空
                notEmpty.await();  // 在notEmpty条件上等待
            return dequeue();      // 出队操作
        } finally {
            lock.unlock();
        }
    }
    
    private void enqueue(E x) {
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length) putIndex = 0;  // 循环数组
        count++;
        notEmpty.signal();  // 唤醒等待的消费者
    }
}

六、实战应用示例

生产者-消费者模式完整实现：

public class ProducerConsumerExample {
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
    
    // 生产者
    class Producer implements Runnable {
        private final String name;
        
        public Producer(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            try {
                for (int i = 0; i < 20; i++) {
                    String item = name + "-" + i;
                    queue.put(item);  // 队列满时自动阻塞
                    System.out.println("生产者 " + name + " 生产: " + item);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    // 消费者
    class Consumer implements Runnable {
        private final String name;
        
        public Consumer(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            try {
                while (true) {
                    String item = queue.take();  // 队列空时自动阻塞
                    System.out.println("消费者 " + name + " 消费: " + item);
                    Thread.sleep(150);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    public void start() {
        // 启动2个生产者，3个消费者
        ExecutorService executor = Executors.newFixedThreadPool(5);
        executor.execute(new Producer("P1"));
        executor.execute(new Producer("P2"));
        executor.execute(new Consumer("C1"));
        executor.execute(new Consumer("C2"));
        executor.execute(new Consumer("C3"));
    }
}

七、性能考量与选型建议

1. ArrayBlockingQueue：固定大小，内存使用可控，适合已知容量的场景
2. LinkedBlockingQueue：可变大小，适合生产消费速度不匹配的场景
3. PriorityBlockingQueue：需要优先级处理的场景
4. SynchronousQueue：直接传递，吞吐量最高，适合任务分发场景

八、常见面试问题

1. BlockingQueue的四种处理方式区别？
2. ArrayBlockingQueue和LinkedBlockingQueue的性能差异？
3. 如何选择合适的BlockingQueue实现？
4. BlockingQueue在线程池中的应用？
5. 如何实现一个自定义的BlockingQueue？

通过以上详细解析，你应该对BlockingQueue有了全面深入的理解，能够在实际开发中正确选择和使用合适的阻塞队列实现。