Java同步队列API怎么用

本文目录导读：

1. 核心特性
2. 创建同步队列
3. 基本操作
4. 实际应用场景
5. 方法总结
6. 注意事项

Java中的同步队列（SynchronousQueue）是java.util.concurrent包下的一种特殊阻塞队列，它的容量为0，每个插入操作必须等待对应的移除操作，以下是详细的使用方法：

核心特性

• 容量为0：不存储任何元素
• 同步机制：每个put必须等待一个take，反之亦然
• 公平性：支持公平和非公平两种模式

创建同步队列

// 默认非公平模式
SynchronousQueue<String> queue = new SynchronousQueue<>();
// 公平模式
SynchronousQueue<String> fairQueue = new SynchronousQueue<>(true);

基本操作

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
public class SynchronousQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        // 创建同步队列
        SynchronousQueue<String> queue = new SynchronousQueue<>();
        // 生产线程
        Thread producer = new Thread(() -> {
            try {
                System.out.println("生产者开始生产...");
                // put会阻塞直到有消费者take
                queue.put("消息1");
                System.out.println("生产者成功放入消息1");
                // 非阻塞尝试，失败立即返回 false
                boolean success = queue.offer("消息2");
                System.out.println("非阻塞尝试放入消息2: " + success);
                // 超时尝试，在指定时间内等待
                success = queue.offer("消息3", 2, TimeUnit.SECONDS);
                System.out.println("超时尝试放入消息3: " + success);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        // 消费线程
        Thread consumer = new Thread(() -> {
            try {
                Thread.sleep(1000); // 延迟1秒
                System.out.println("消费者开始消费...");
                // take会阻塞直到有元素可用
                String message = queue.take();
                System.out.println("消费者收到消息: " + message);
                // 非阻塞尝试
                message = queue.poll();
                System.out.println("非阻塞尝试获取: " + message);
                // 超时尝试
                message = queue.poll(2, TimeUnit.SECONDS);
                System.out.println("超时尝试获取: " + message);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        producer.start();
        consumer.start();
    }
}

实际应用场景

1 生产者-消费者模式

import java.util.concurrent.SynchronousQueue;
public class ProducerConsumerExample {
    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        // 生产者
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    System.out.println("生产数据: " + i);
                    queue.put(i);
                    System.out.println("数据 " + i + " 已交付");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        // 消费者
        Thread consumer = new Thread(() -> {
            try {
                Integer data;
                while ((data = queue.poll(3, TimeUnit.SECONDS)) != null) {
                    System.out.println("消费数据: " + data);
                    Thread.sleep(100); // 模拟处理时间
                }
                System.out.println("消费者停止消费");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        producer.start();
        consumer.start();
    }
}

2 线程池的实现

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolWithSynchronousQueue {
    public static void main(String[] args) {
        // 使用SynchronousQueue的线程池，当提交任务时如果没有空闲线程则创建新线程
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            1,                // 核心线程数
            10,               // 最大线程数
            60,               // 空闲线程存活时间
            TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>()  // 使用同步队列
        );
        // 提交任务
        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println("执行任务 " + taskId + " by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
    }
}

方法总结

注意事项

• 容量为0：不能用于存储元素
• 线程配对：生产者和消费者必须同时准备好
• 性能优化：在高并发场景下效率较高
• 适用场景：适用于经典的握手、传输数据等场景

SynchronousQueue特别适合在需要直接传递任务而不需要缓冲的场景中使用,比如某些线程池的实现中。