Likanug // lab
返回文章列表

文章

技术分享

关于阻塞队列

从API 概览、常见实现、使用技巧到典型应用场景说明阻塞队列

#Java#并发编程
目录
  1. 下面从API 概览、常见实现、使用技巧到典型应用场景系统地聊聊 java.util.concurrent.BlockingQueue。
  2. 1. API 概览
  3. 2. 常见实现类
  4. 3. 使用技巧
  5. 4. 典型应用场景
  6. 5. 快速示例：生产者–消费者
  7. 小结
  8. 📎 参考文章

下面从API 概览、常见实现、使用技巧到典型应用场景系统地聊聊 java.util.concurrent.BlockingQueue。#

1. API 概览#

BlockingQueue<E> 是一个支持阻塞读写的队列接口。常用方法分为四类，每类各四个变体（抛异常/返回特殊值/超时/无限阻塞）：

语义抛异常返回值超时阻塞（无限）
**插入**`add(e)``offer(e)``offer(e,t,unit)``put(e)`
**移除**`remove()``poll()``poll(t,unit)``take()`
**查看头部**`element()``peek()`——
  • 阻塞/无限等待：put() 在队满时阻塞，take() 在队空时阻塞。
  • 带超时：offer(e,1,TimeUnit.SECONDS) 最多等 1s，否则返回 false；poll(1,TimeUnit.SECONDS) 返回 null。
  • 非阻塞失败安全：offer/poll 不会抛异常，用于不想阻塞的场景。

2. 常见实现类#

类名类型特点
`ArrayBlockingQueue`有界队列底层数组，固定容量，构造时可指定公平性（FIFO）
`LinkedBlockingQueue`可选有界/无界链表实现，吞吐高；默认无界（`Integer.MAX_VALUE`）
`SynchronousQueue`同步队列无缓冲，生产者 `put` 必须等待消费者 `take`
`PriorityBlockingQueue`优先级队列无界，自动按元素自然顺序或自定义 `Comparator` 排序
`DelayQueue`延时队列元素需实现 `Delayed`，到期后才能取出，常用于定时任务
`LinkedTransferQueue`转移队列支持 `transfer` 操作，让消费者立即消费，否则等队列消费

3. 使用技巧#

  1. 控制队列长度
    • 对于 LinkedBlockingQueue 或 ArrayBlockingQueue，应显式指定容量，避免默认无界队列 OOM。
    • 使用带超时的 offer/poll 可实现“背压”（生产者不会无限制等待，也不会无限入队）。
  2. 公平性 vs 性能
    • ArrayBlockingQueue 构造时可以开公平锁：new ArrayBlockingQueue<>(capacity, true)。
    • 公平模式下线程按 FIFO 排队竞争，吞吐稍逊；非公平模式性能更高。
  3. 批量操作
    • drainTo(Collection<? super E> c)：一次性将队列中元素转移到另一个集合，减少多次 poll() 的锁开销。
  4. 监控队列状态
    • remainingCapacity()、size()、peek() 等方法可用于监控和调优；注意并发下的瞬时性。
  5. 优雅关闭
    • 生产者在结束时可向队列发送“结束标志”（比如 null（需特殊处理）或自定义哨兵对象），消费者收到后跳出循环。

4. 典型应用场景#

场景推荐实现原理/说明
**生产者–消费者**`ArrayBlockingQueue`固定缓冲区，可控制背压；生产者 `put()`，消费者 `take()`
**线程池工作队列**`LinkedBlockingQueue`无界或有界队列作任务缓冲，配合 `ThreadPoolExecutor`
**流水线分阶段处理**`SynchronousQueue`阶段间零缓冲交替，生产者/消费者一对一实时交接
**优先级任务调度**`PriorityBlockingQueue`可根据任务优先级排序执行
**延时任务调度**`DelayQueue`按到期时间释放元素，配合专门的调度线程实现延时执行
**异步日志/IO 写入**`LinkedTransferQueue`高吞吐、低延迟，`transfer` 保证直接交付或让调用者等待消费者处理
**限流与散流**`Semaphore` + `BlockingQueue`结合信号量和有界队列，可实现在高并发下对请求做削峰填谷

5. 快速示例：生产者–消费者#

import java.util.concurrent.*;

public class ProducerConsumerDemo {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

        // 生产者
        Runnable producer = () -> {
            try {
                for (int i = 0; i < 20; i++) {
                    queue.put("msg-" + i);
                    System.out.println("Produced msg-" + i);
                }
                queue.put("DONE"); // 结束标志
            } catch (InterruptedException ignored) {}
        };

        // 消费者
        Runnable consumer = () -> {
            try {
                String msg;
                while (!"DONE".equals(msg = queue.take())) {
                    System.out.println("Consumed " + msg);
                }
                System.out.println("Consumer exiting");
            } catch (InterruptedException ignored) {}
        };

        new Thread(producer).start();
        new Thread(consumer).start();
    }
}

小结#

  • 选对实现：根据并发量、缓冲需求、是否有优先级/延时等选合适队列；
  • 合理设容量：避免无界队列内存爆炸，或太小导致过度阻塞；
  • 使用超时：在生产者/消费者中用带超时的 offer/poll 实现更灵活的背压；
  • 监控与批量：定时 drainTo、监控 remainingCapacity 便于性能调优；
  • 合并其他工具：可和 Semaphore、LockSupport、Phaser 等组合，满足更复杂的并发场景。 掌握好这些方法和技巧，就能在多线程架构中高效、稳健地使用阻塞队列。

📎 参考文章#