Java 阻塞队列:7种类型全解析

发布于:2025-07-09 ⋅ 阅读:(15) ⋅ 点赞:(0)

在 Java 中,阻塞队列(BlockingQueue) 是一种线程安全的队列结构,用于实现生产者-消费者模式。它的核心特性是在队列为空时阻塞消费者线程,在队列满时阻塞生产者线程,从而自动协调线程之间的协作。


阻塞队列的核心特性

  1. 线程安全:所有操作(如 puttake)都是线程安全的。
  2. 阻塞操作
    • 队列满时:生产者线程阻塞(等待消费者消费)。
    • 队列空时:消费者线程阻塞(等待生产者生产)。
  3. 超时控制:支持带超时时间的操作(如 offer(timeout, unit)poll(timeout, unit))。
  4. 公平策略:部分实现(如 ArrayBlockingQueue)支持公平锁,防止线程饥饿。

Java 中的 7 种阻塞队列

以下是 Java 提供的 7 种阻塞队列及其特点和适用场景:

1. ArrayBlockingQueue(有界队列)
  • 特点
    • 基于数组实现,容量固定(初始化时指定)。
    • 支持公平锁(默认非公平锁)。
  • 适用场景
    • 需要严格限制队列容量的场景(如任务量可控的线程池)。
  • 示例代码
    BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
    queue.put("A"); // 队列满时阻塞
    String task = queue.take(); // 队列空时阻塞
    
2. LinkedBlockingQueue(有界/无界队列)
  • 特点
    • 基于链表实现,默认容量为 Integer.MAX_VALUE(可视为无界)。
    • 入队和出队使用独立锁(putLocktakeLock),提高并发性能。
  • 适用场景
    • 任务量不可预测的高吞吐量场景(如线程池默认队列)。
  • 示例代码
    BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(100); // 指定容量
    queue.put(1); // 队列满时阻塞
    Integer task = queue.take(); // 队列空时阻塞
    
3. PriorityBlockingQueue(无界优先级队列)
  • 特点
    • 无界队列,基于优先级堆实现。
    • 元素必须实现 Comparable 接口或提供比较器。
  • 适用场景
    • 需要按优先级处理任务(如紧急任务优先)。
  • 示例代码
    BlockingQueue<PriorityTask> queue = new PriorityBlockingQueue<>();
    queue.put(new PriorityTask(1)); // 任务优先级由 compareTo() 决定
    PriorityTask task = queue.take();
    
4. DelayQueue(延迟队列)
  • 特点
    • 无界队列,元素必须实现 Delayed 接口。
    • 只有在延迟时间到达后,任务才能被取出。
  • 适用场景
    • 定时任务(如缓存过期、订单超时)。
  • 示例代码
    DelayQueue<DelayedTask> queue = new DelayQueue<>();
    queue.put(new DelayedTask(5000)); // 5 秒后可用
    DelayedTask task = queue.take(); // 等待任务延迟时间到期
    
5. SynchronousQueue(同步队列)
  • 特点
    • 无容量队列,每个插入操作必须等待一个对应的移除操作。
    • 生产者和消费者直接传递数据。
  • 适用场景
    • 高吞吐量的短任务场景(如 newCachedThreadPool)。
  • 示例代码
    BlockingQueue<String> queue = new SynchronousQueue<>();
    queue.put("X"); // 阻塞直到有消费者调用 take()
    String task = queue.take(); // 阻塞直到有生产者调用 put()
    
6. LinkedTransferQueue(无界队列)
  • 特点
    • 支持“预占模式”(生产者和消费者直接交互)。
    • 高性能的无界队列。
  • 适用场景
    • 需要高效传递任务的场景(如快速响应的系统)。
  • 示例代码
    BlockingQueue<Integer> queue = new LinkedTransferQueue<>();
    queue.put(100); // 直接传递给消费者
    Integer task = queue.take(); // 立即获取任务
    
7. LinkedBlockingDeque(双向阻塞队列)
  • 特点
    • 双端队列,支持从两端插入/移除元素。
    • 可作为有界或无界队列。
  • 适用场景
    • 工作窃取算法(如 ForkJoinPool)。
  • 示例代码
    BlockingQueue<String> queue = new LinkedBlockingDeque<>(100);
    queue.put("A"); // 从尾部插入
    String task = queue.take(); // 从头部移除
    

阻塞队列的核心方法

方法 行为 抛出异常 返回布尔值 阻塞 超时阻塞
add(E e) 插入元素 队列满时抛异常 -
offer(E e) 插入元素 队列满时返回 false -
offer(E e, long timeout, TimeUnit unit) 插入元素 超时后返回 false -
put(E e) 插入元素 - -
remove() 移除元素 队列空时抛异常 -
poll() 移除元素 队列空时返回 null -
poll(long timeout, TimeUnit unit) 移除元素 超时后返回 null -
take() 移除元素 - -

如何选择阻塞队列?

  1. 有界 vs 无界
    • 有界队列(如 ArrayBlockingQueue):防止内存溢出,需合理设置容量。
    • 无界队列(如 LinkedBlockingQueue):可能导致任务堆积,需结合拒绝策略使用。
  2. 优先级需求
    • 使用 PriorityBlockingQueue 实现优先级排序。
  3. 延迟任务
    • 使用 DelayQueue 实现定时或延迟执行。
  4. 高性能场景
    • SynchronousQueue 适合高吞吐量的短任务。
    • LinkedTransferQueue 适合快速传递任务的场景。

示例:使用阻塞队列实现生产者-消费者模型

import java.util.concurrent.*;

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

        // 生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    String task = "Task-" + i;
                    queue.put(task); // 队列满时阻塞
                    System.out.println("Produced: " + task);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String task = queue.take(); // 队列空时阻塞
                    System.out.println("Consumed: " + task);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

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

注意事项

  1. 避免内存泄漏
    • 使用无界队列时,需结合线程池的拒绝策略(如 CallerRunsPolicy)。
  2. 公平性
    • ArrayBlockingQueue 支持公平锁(构造函数传入 true)。
  3. 线程中断
    • 阻塞操作(如 put/take)会响应中断,需捕获 InterruptedException

通过合理选择阻塞队列类型,可以高效地实现线程间的协作,解决生产者-消费者问题。


网站公告

今日签到

点亮在社区的每一天
去签到