在Java中如何使用BlockingQueue实现线程安全队列_BlockingQueue线程同步技巧说明-java教程-PHP中文网

在Java中如何使用BlockingQueue实现线程安全队列_BlockingQueue线程同步技巧说明

P粉602998670

发布： 2025-11-20 12:01:05

原创

364人浏览过

BlockingQueue是Java中用于线程安全队列操作的接口，支持阻塞式生产者-消费者模型。1. 核心特性：队列空时take()阻塞，满时put()阻塞，自动同步无需显式加锁。2. 常用实现：ArrayBlockingQueue（有界数组）、LinkedBlockingQueue（可选有界链表）、PriorityBlockingQueue（优先级排序）、SynchronousQueue（不存储元素）。3. 关键方法：add()/offer()/put()控制入队行为，poll()/take()控制出队，支持超时与非阻塞操作。4. 示例中通过LinkedBlockingQueue实现生产者每500ms生产消息，消费者每1s消费，直至收到exit信号。5. 使用时需避免null插入，合理设置容量防内存溢出，并处理中断异常。

在java中如何使用blockingqueue实现线程安全队列_blockingqueue线程同步技巧说明

在Java中，BlockingQueue 是一个支持线程安全的队列接口，常用于生产者-消费者模型中的线程同步。它位于 java.util.concurrent 包中，所有实现都具备自动阻塞能力，能有效避免多线程环境下的竞态条件。

BlockingQueue 核心特性

BlockingQueue 的最大特点是：当队列为空时，从队列获取元素的线程会阻塞；当队列为满时，向队列插入元素的线程也会阻塞。这种机制天然适合线程间协作。

线程安全：无需手动加锁，内部已实现同步
阻塞操作：take() 和 put() 方法会自动阻塞等待
支持超时：offer(e, time) 和 poll(time) 可设置等待时限
不可为 null：任何添加 null 元素的操作都会抛出 NullPointerException

常用实现类介绍

根据使用场景不同，可以选择不同的实现：

ArrayBlockingQueue：基于数组的有界阻塞队列，需指定容量
LinkedBlockingQueue：基于链表的可选有界队列，默认容量为 Integer.MAX_VALUE
PriorityBlockingQueue：支持优先级排序的无界阻塞队列
SynchronousQueue：不存储元素的阻塞队列，每个插入必须等待对应移除

生产者-消费者示例代码

以下是一个简单的生产者-消费者模型演示：

OmniAudio

OmniAudio 是一款通过 AI 支持将网页、Word 文档、Gmail 内容、文本片段、视频音频文件都转换为音频播客，并生成可在常见 Podcast ap

111

查看详情

立即学习“Java免费学习笔记（深入）”；

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

// 消息类
class Message {
    private final String content;
    public Message(String content) {
        this.content = content;
    }
    public String toString() {
        return "Message: " + content;
    }
}

// 生产者
class Producer implements Runnable {
    private final BlockingQueue<Message> queue;
    public Producer(BlockingQueue<Message> queue) {
        this.queue = queue;
    }

    public void run() {
        try {
            for (int i = 1; i <= 5; i++) {
                Message msg = new Message("data-" + i);
                queue.put(msg); // 自动阻塞，直到空间可用
                System.out.println("Produced " + msg);
                Thread.sleep(500); // 模拟耗时
            }
            queue.put(new Message("exit")); // 发送结束信号
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 消费者
class Consumer implements Runnable {
    private final BlockingQueue<Message> queue;
    public Consumer(BlockingQueue<Message> queue) {
        this.queue = queue;
    }

    public void run() {
        try {
            Message msg;
            while ((msg = queue.take()) != null) {
                System.out.println("Consumed " + msg);
                if ("exit".equals(msg.toString())) break;
                Thread.sleep(1000); // 模拟处理时间
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 主程序
public class BlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<Message> queue = new LinkedBlockingQueue<>(10);

        Thread producer = new Thread(new Producer(queue));
        Thread consumer = new Thread(new Consumer(queue));

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

        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

登录后复制