本文目录导读：
我来为您提供一个完整的BlockingQueue案例,涵盖生产者-消费者模式的实际应用。
基础案例：生产者-消费者模式
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public class BlockingQueueBasicExample {
public static void main(String[] args) {
// 创建一个容量为3的阻塞队列
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(3);
// 创建生产者线程
Thread producer = new Thread(() -> {
try {
int value = 0;
while (true) {
value++;
// put方法会阻塞直到队列有空间
queue.put(value);
System.out.println("生产者生产: " + value + ", 队列大小: " + queue.size());
Thread.sleep(500);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
// 创建消费者线程
Thread consumer = new Thread(() -> {
try {
while (true) {
// take方法会阻塞直到队列有元素
Integer value = queue.take();
System.out.println("消费者消费: " + value);
Thread.sleep(1000);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
}
}
完整的生产者-消费者案例（多生产多消费）
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class MultiProducerConsumerExample {
// 任务队列
private static final BlockingQueue<Task> taskQueue = new ArrayBlockingQueue<>(10);
private static final AtomicInteger taskIdGenerator = new AtomicInteger(0);
private static final AtomicInteger completedTasks = new AtomicInteger(0);
// 任务类
static class Task {
private final int id;
private final String name;
public Task(int id, String name) {
this.id = id;
this.name = name;
}
public int getId() { return id; }
public String getName() { return name; }
@Override
public String toString() {
return "Task{id=" + id + ", name='" + name + "'}";
}
}
// 生产者
static class Producer implements Runnable {
private final String producerName;
private final int taskCount;
public Producer(String name, int taskCount) {
this.producerName = name;
this.taskCount = taskCount;
}
@Override
public void run() {
try {
for (int i = 0; i < taskCount; i++) {
int taskId = taskIdGenerator.incrementAndGet();
Task task = new Task(taskId, "Task-" + taskId);
// offer with timeout，避免无限阻塞
boolean added = taskQueue.offer(task, 2, TimeUnit.SECONDS);
if (added) {
System.out.println(producerName + " 生产了: " + task);
} else {
System.out.println(producerName + " 队列已满，任务被丢弃: " + task);
}
Thread.sleep(100);
}
System.out.println(producerName + " 完成生产任务");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println(producerName + " 被中断");
}
}
}
// 消费者
static class Consumer implements Runnable {
private final String consumerName;
private final int maxTasks;
public Consumer(String name, int maxTasks) {
this.consumerName = name;
this.maxTasks = maxTasks;
}
@Override
public void run() {
try {
while (completedTasks.get() < maxTasks && !Thread.currentThread().isInterrupted()) {
// poll with timeout，避免无限等待
Task task = taskQueue.poll(2, TimeUnit.SECONDS);
if (task != null) {
System.out.println(consumerName + " 消费了: " + task);
completedTasks.incrementAndGet();
Thread.sleep(200);
} else {
System.out.println(consumerName + " 队列为空，等待中...");
Thread.sleep(500);
}
}
System.out.println(consumerName + " 完成消费任务");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println(consumerName + " 被中断");
}
}
}
public static void main(String[] args) throws InterruptedException {
int producerCount = 2;
int consumerCount = 3;
int tasksPerProducer = 5;
System.out.println("=== 开始执行生产者消费者模式 ===");
System.out.println("生产者数量: " + producerCount + ", 消费者数量: " + consumerCount);
System.out.println("每个生产者生产任务数: " + tasksPerProducer + ", 总任务数: " + (producerCount * tasksPerProducer));
// 启动生产者
for (int i = 0; i < producerCount; i++) {
new Thread(new Producer("生产者-" + (i + 1), tasksPerProducer)).start();
}
// 启动消费者
for (int i = 0; i < consumerCount; i++) {
new Thread(new Consumer("消费者-" + (i + 1), producerCount * tasksPerProducer)).start();
}
// 主线程等待一段时间后退出
Thread.sleep(10000);
System.out.println("=== 执行完成 ===");
System.out.println("已完成任务数: " + completedTasks.get());
System.out.println("队列剩余任务数: " + taskQueue.size());
}
}
带优先级的生产者-消费者案例
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
public class PriorityQueueExample {
static class WorkOrder implements Comparable<WorkOrder> {
private final int id;
private final int priority; // 1-高优先级, 2-中优先级, 3-低优先级
private final String description;
public WorkOrder(int id, int priority, String description) {
this.id = id;
this.priority = priority;
this.description = description;
}
@Override
public int compareTo(WorkOrder other) {
return Integer.compare(this.priority, other.priority);
}
@Override
public String toString() {
return String.format("WorkOrder{id=%d, priority=%s, desc='%s'}",
id, priority == 1 ? "高" : priority == 2 ? "中" : "低", description);
}
}
public static void main(String[] args) throws InterruptedException {
// 使用优先级阻塞队列
BlockingQueue<WorkOrder> queue = new PriorityBlockingQueue<>();
// 生产者线程
Thread producer = new Thread(() -> {
try {
int id = 1;
String[] descriptions = {"处理订单", "紧急维修", "日常检查", "数据备份", "系统升级"};
int[] priorities = {3, 1, 2, 3, 1};
for (int i = 0; i < descriptions.length; i++) {
WorkOrder order = new WorkOrder(id++, priorities[i], descriptions[i]);
queue.put(order);
System.out.println(Thread.currentThread().getName() + " 添加: " + order);
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "生产者线程");
// 消费者线程
Thread consumer = new Thread(() -> {
try {
while (!Thread.currentThread().isInterrupted()) {
WorkOrder order = queue.poll(100, TimeUnit.MILLISECONDS);
if (order != null) {
System.out.println(Thread.currentThread().getName() + " 处理: " + order);
Thread.sleep(200);
} else {
break;
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "消费者线程");
producer.start();
consumer.start();
producer.join();
consumer.join(5000);
System.out.println("队列中剩余任务: " + queue.size());
}
}
延迟队列案例（定时任务处理）
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class DelayQueueExample {
static class ScheduledTask implements Delayed {
private final String taskName;
private final long executeTime;
public ScheduledTask(String taskName, long delayInMillis) {
this.taskName = taskName;
this.executeTime = System.currentTimeMillis() + delayInMillis;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed other) {
return Long.compare(this.executeTime, ((ScheduledTask) other).executeTime);
}
@Override
public String toString() {
return "ScheduledTask{名称='" + taskName + "', 计划执行时间=" +
(executeTime - System.currentTimeMillis()) + "ms后}";
}
}
public static void main(String[] args) throws InterruptedException {
BlockingQueue<ScheduledTask> delayQueue = new DelayQueue<>();
// 添加定时任务
delayQueue.put(new ScheduledTask("任务A", 5000)); // 5秒后执行
delayQueue.put(new ScheduledTask("任务B", 3000)); // 3秒后执行
delayQueue.put(new ScheduledTask("任务C", 1000)); // 1秒后执行
System.out.println("已添加3个定时任务");
// 消费任务
while (!delayQueue.isEmpty()) {
System.out.println("等待取任务，队列大小: " + delayQueue.size());
ScheduledTask task = delayQueue.take();
System.out.println("执行: " + task);
}
System.out.println("所有定时任务执行完毕");
}
}
实际业务场景：订单处理系统
import java.util.concurrent.*;
public class OrderProcessingSystem {
// 订单类
static class Order {
private final String orderId;
private final String customerName;
private final double amount;
private final long timestamp;
public Order(String orderId, String customerName, double amount) {
this.orderId = orderId;
this.customerName = customerName;
this.amount = amount;
this.timestamp = System.currentTimeMillis();
}
@Override
public String toString() {
return String.format("订单{ID=%s, 客户=%s, 金额=%.2f}",
orderId, customerName, amount);
}
}
// 订单处理器
static class OrderProcessor {
private final BlockingQueue<Order> orderQueue;
private final BlockingQueue<Order> processedOrders;
public OrderProcessor(int queueCapacity) {
this.orderQueue = new ArrayBlockingQueue<>(queueCapacity);
this.processedOrders = new ArrayBlockingQueue<>(queueCapacity);
}
// 接收订单
public boolean receiveOrder(Order order) throws InterruptedException {
return orderQueue.offer(order, 5, TimeUnit.SECONDS);
}
// 处理订单线程
public void startOrderProcessing() {
// 处理线程
Thread processingThread = new Thread(() -> {
try {
while (true) {
Order order = orderQueue.poll(3, TimeUnit.SECONDS);
if (order == null) {
System.out.println("[处理器] 暂无待处理订单...");
continue;
}
System.out.println("[处理器] 正在处理: " + order);
// 模拟处理时间
Thread.sleep(500);
// 模拟处理成功
processedOrders.put(order);
System.out.println("[处理器] 处理完成: " + order);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
processingThread.start();
}
// 获取已处理订单
public void startOrderMonitoring() {
Thread monitorThread = new Thread(() -> {
try {
while (true) {
Order processed = processedOrders.poll(1, TimeUnit.SECONDS);
if (processed != null) {
System.out.println("[监控] 订单已发货: " + processed);
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
monitorThread.start();
}
}
public static void main(String[] args) throws InterruptedException {
System.out.println("=== 订单处理系统启动 ===");
OrderProcessor orderProcessor = new OrderProcessor(10);
orderProcessor.startOrderProcessing();
orderProcessor.startOrderMonitoring();
// 模拟多个客户提交订单
for (int i = 0; i < 20; i++) {
Order order = new Order(
"ORD-" + String.format("%04d", i),
"客户-" + (i + 1),
100 + Math.random() * 900
);
boolean accepted = orderProcessor.receiveOrder(order);
if (accepted) {
System.out.println("[系统] 已接收: " + order);
} else {
System.out.println("[系统] 队列已满，拒绝: " + order);
}
Thread.sleep(200);
}
// 运行一段时间后停止
Thread.sleep(5000);
System.out.println("=== 订单处理系统停止 ===");
System.exit(0);
}
}
关键要点说明
- ArrayBlockingQueue: 基于数组的有界阻塞队列，FIFO顺序
- PriorityBlockingQueue: 带优先级的无界阻塞队列
- DelayQueue: 延迟队列，元素有延迟时间
- put/take: 阻塞操作，会等待
- offer/poll: 非阻塞或超时操作
- BlockingQueue 常用于：
- 生产者-消费者模式
- 线程池任务队列
- 消息缓冲
- 订单处理系统
- 多级流水线处理
这些案例展示了BlockingQueue在实际开发中的典型应用场景,可以根据具体需求选择合适的队列类型和操作方法。
