Java 阻塞队列 (BlockingQueue)：从原理本质到生产实战

java.util.Collection
└── java.util.Queue
    └── java.util.concurrent.BlockingQueue
        ├── ArrayBlockingQueue          // 有界，数组实现
        ├── LinkedBlockingQueue         // 可选有界，链表实现
        ├── PriorityBlockingQueue       // 无界，支持优先级
        ├── DelayQueue                  // 无界，延迟元素
        ├── SynchronousQueue            // 容量为0，不存储元素
        ├── LinkedTransferQueue         // 无界，支持转移
        └── LinkedBlockingDeque         // 双端阻塞队列

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 演示：基本的生产者-消费者模型
 * 重点：如何使用“毒丸对象”让消费者优雅退出
 */
public class BasicProducerConsumer {

    // 1. 创建有界阻塞队列，容量为 5
    private static final BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
    
    // 2. 定义“毒丸”常量，用于通知停止
    private static final String POISON_PILL = "POISON_PILL";

    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== 启动系统 ===");

        // 启动消费者
        Thread consumer = new Thread(new Consumer(), "Consumer-Thread");
        consumer.start();

        // 启动生产者
        Thread producer = new Thread(new Producer(), "Producer-Thread");
        producer.start();

        // 等待生产者结束
        producer.join();
        
        // 给消费者一点时间处理剩余消息
        TimeUnit.SECONDS.sleep(1);
        
        System.out.println("=== 系统关闭 ===");
    }

    /**
     * 生产者：生产 10 条消息，最后发送毒丸
     */
    static class Producer implements Runnable {
        @Override
        public void run() {
            try {
                for (int i = 1; i <= 10; i++) {
                    String msg = "Message-" + i;
                    // put()：如果队列满，线程在此阻塞等待
                    queue.put(msg);
                    System.out.println(Thread.currentThread().getName() + " 生产: " + msg);
                    
                    // 模拟生产速度略快于消费，触发阻塞测试
                    TimeUnit.MILLISECONDS.sleep(50);
                }
                
                // 生产完毕，放入毒丸
                queue.put(POISON_PILL);
                System.out.println(Thread.currentThread().getName() + " 生产结束，发送停止信号。");
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 消费者：持续消费，直到读到毒丸
     */
    static class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    // take()：如果队列空，线程在此阻塞等待
                    String msg = queue.take();
                    
                    // 检查是否是毒丸
                    if (POISON_PILL.equals(msg)) {
                        System.out.println(Thread.currentThread().getName() + " 收到停止信号，退出。");
                        break; // 跳出循环，线程自然结束
                    }
                    
                    // 模拟业务处理
                    TimeUnit.MILLISECONDS.sleep(100);
                    System.out.println(Thread.currentThread().getName() + " 消费: " + msg);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

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

/**
 * 演示：生产环境推荐的用法
 * 重点：使用 offer/poll 带超时，以及响应 interrupt
 */
public class AdvancedBlockingQueueDemo {

    // 使用 LinkedBlockingQueue，但必须指定容量！防止 OOM
    private static final BlockingQueue<String> queue = new LinkedBlockingQueue<>(10);
    
    // 控制标志
    private static volatile boolean running = true;

    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== 启动高级演示 ===");

        Thread producer = new Thread(new AdvProducer(), "Adv-Producer");
        Thread consumer = new Thread(new AdvConsumer(), "Adv-Consumer");

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

        // 模拟运行 3 秒
        TimeUnit.SECONDS.sleep(3);

        System.out.println("=== 主线程发起关闭 ===");
        running = false;
        
        // 关键：中断工作线程，使它们从阻塞中醒来
        producer.interrupt();
        consumer.interrupt();

        producer.join();
        consumer.join();
        
        System.out.println("=== 所有线程已安全退出 ===");
    }

    static class AdvProducer implements Runnable {
        @Override
        public void run() {
            int count = 0;
            while (running) {
                try {
                    String item = "Item-" + (++count);
                    
                    // 【推荐】offer 带超时：最多等 1 秒，如果还满则放弃或重试
                    boolean success = queue.offer(item, 1, TimeUnit.SECONDS);
                    
                    if (success) {
                        System.out.println(Thread.currentThread().getName() + " 放入: " + item);
                    } else {
                        System.out.println(Thread.currentThread().getName() + " 队列满，丢弃: " + item);
                    }
                    
                    TimeUnit.MILLISECONDS.sleep(50);

                } catch (InterruptedException e) {
                    System.out.println(Thread.currentThread().getName() + " 被中断，停止生产。");
                    Thread.currentThread().interrupt(); // 恢复中断状态
                    break;
                }
            }
        }
    }

    static class AdvConsumer implements Runnable {
        @Override
        public void run() {
            // 即使 running=false，也要把队列里剩下的处理完
            while (running || !queue.isEmpty()) {
                try {
                    // 【推荐】poll 带超时：最多等 1 秒，如果还空则返回 null
                    String item = queue.poll(1, TimeUnit.SECONDS);
                    
                    if (item != null) {
                        System.out.println(Thread.currentThread().getName() + " 取出: " + item);
                        // 处理业务...
                    } else {
                        // 超时返回 null，说明暂时没数据，可以做一些心跳或轻量检查
                        // System.out.println("等待数据...");
                    }

                } catch (InterruptedException e) {
                    System.out.println(Thread.currentThread().getName() + " 被中断，停止消费。");
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
}