在Java中如何使用同步队列SynchronousQueue-java教程-PHP中文网

在Java中如何使用同步队列SynchronousQueue

P粉602998670

发布： 2025-10-07 11:18:02

原创

659人浏览过

SynchronousQueue不存储元素，每个put操作需等待take操作完成，适用于生产者消费者直接传递数据的场景。1. 它无内部容量，插入和移除必须同步配对。2. 不支持null值，不能遍历。3. 常用于newCachedThreadPool线程池任务传递。4. 提供offer/poll等超时方法实现限时交互。5. 适合一对一同步传递，不适合缓冲或批量处理。示例中生产者调用put后阻塞，直到消费者调用take完成“握手”，数据直接传递。注意其性能依赖线程调度，调试困难因无中间状态。核心是理解其“只传递、不存储”的设计本质。

在java中如何使用同步队列synchronousqueue

SynchronousQueue 是 Java 并发包 java.util.concurrent 中的一个特殊阻塞队列，它没有内部容量，每个插入操作必须等待另一个线程的对应移除操作，反之亦然。它适用于在生产者和消费者之间直接传递数据的场景，常用于线程池如 Executors.newCachedThreadPool() 中。

理解 SynchronousQueue 的工作方式

SynchronousQueue 不存储元素。当你调用 put() 方法时，该方法会阻塞，直到另一个线程调用 take() 来接收这个元素；同样，如果一个线程尝试 take()，它会一直等待直到有另一个线程执行 put()。这种“握手”机制实现了线程间直接的数据传递。

主要特点：

不支持 null 值
不能遍历，因为没有存储空间
非常适合工作窃取或高并发任务传递场景

基本使用示例：生产者-消费者模式

下面是一个简单的生产者和消费者通过 SynchronousQueue 通信的例子：

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

import java.util.concurrent.SynchronousQueue;

public class SynchronousQueueExample {
    public static void main(String[] args) {
        SynchronousQueue<String> queue = new SynchronousQueue<>();

        // 生产者线程
        new Thread(() -> {
            try {
                System.out.println("生产者准备放入数据");
                queue.put("Hello from producer");
                System.out.println("数据已放入");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();

        // 消费者线程
        new Thread(() -> {
            try {
                System.out.println("消费者等待数据");
                String data = queue.take();
                System.out.println("收到数据: " + data);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

登录后复制

运行结果会显示两个线程“握手”成功，数据从生产者直接传递给消费者。

如知AI笔记

如知笔记——支持markdown的在线笔记，支持ai智能写作、AI搜索，支持DeepseekR1满血大模型

查看详情

使用 offer 和 poll 进行非阻塞或限时操作

除了阻塞的 put/take，SynchronousQueue 还提供非阻塞或超时版本的方法：

offer(E e)：尝试将元素放入队列，如果没有消费者等待，立即返回 false
offer(E e, long timeout, TimeUnit unit)：最多等待指定时间，如果有消费者在此期间到来并接收，则返回 true
poll(long timeout, TimeUnit unit)：尝试获取元素，最多等待指定时间

示例：带超时的交互

new Thread(() -> {
    try {
        boolean success = queue.offer("Timeout data", 2, TimeUnit.SECONDS);
        if (success) {
            System.out.println("数据发送成功");
        } else {
            System.out.println("发送超时，无消费者接收");
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

new Thread(() -> {
    try {
        Thread.sleep(3000); // 故意延迟，使 offer 超时
        String data = queue.poll(5, TimeUnit.SECONDS);
        System.out.println("消费者最终收到: " + data);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

登录后复制