Java中如何使用BlockingQueue实现线程安全队列_技术教程

BlockingQueue是Java中用于线程安全数据传递的阻塞队列，支持生产者-消费者模式。其核心特性为：队列为空时取元素阻塞，为满时插入元素阻塞。主要方法包括put()、take()、offer()带超时和poll()带超时。常见实现有ArrayBlockingQueue（有界数组）、LinkedBlockingQueue（可选有界链表）、SynchronousQueue（不存储元素）和PriorityBlockingQueue（优先级排序）。示例中使用ArrayBlockingQueue实现生产者发送10条消息并发送END信号，消费者接收并处理消息，收到END后退出。使用建议包括根据场景选择合适实现、避免无界队列导致内存溢出、正确处理中断异常，并利用阻塞机制简化线程协作。

在Java中，BlockingQueue 是一个支持阻塞插入和移除操作的线程安全队列。它位于 java.util.concurrent 包下，非常适合用于生产者-消费者模型中，多个线程之间安全地传递数据。

BlockingQueue 的基本特性

BlockingQueue 的核心特点是：当队列为空时，尝试获取元素的线程会被阻塞；当队列为满时，尝试添加元素的线程也会被阻塞。这种机制天然支持线程间的协调。

主要方法包括：

put(E e)：将元素插入队列，如果队列满，则阻塞等待空间。
take()：从队列中取出一个元素，如果队列为空，则阻塞等待元素。
offer(E e, long timeout, TimeUnit unit)：带超时的插入。
poll(long timeout, TimeUnit unit)：带超时的取出。

常用实现类

BlockingQueue 是接口，常见的实现有：

ArrayBlockingQueue：基于数组的有界阻塞队列，需指定容量。
LinkedBlockingQueue：基于链表的可选有界队列，默认为无界（但建议设上限）。
SynchronousQueue：不存储元素的阻塞队列，每个插入必须等待另一个线程的移除。
PriorityBlockingQueue：支持优先级排序的无界阻塞队列。

生产者-消费者示例

下面是一个使用 ArrayBlockingQueue 实现生产者和消费者线程安全通信的例子：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;

public class ProducerConsumerExample {
    private static final int QUEUE_CAPACITY = 5;

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

        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    String item = "消息-" + i;
                    queue.put(item);
                    System.out.println("生产者发送: " + item);
                    Thread.sleep(500); // 模拟耗时
                }
                queue.put("END"); // 发送结束信号
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String item = queue.take();
                    if ("END".equals(item)) {
                        System.out.println("消费者收到结束信号，退出。");
                        break;
                    }
                    System.out.println("消费者收到: " + item);
                    Thread.sleep(800); // 模拟处理时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

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