BlockingQueue案例

wen java案例 2026-08-09 218

本文目录导读：

BlockingQueue案例

基础案例：生产者-消费者模式
完整的生产者-消费者案例（多生产多消费）
带优先级的生产者-消费者案例
延迟队列案例（定时任务处理）
实际业务场景：订单处理系统
关键要点说明

我来为您提供一个完整的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);
    }
}