Java中的LinkedBlockingQueue：原理、应用与性能深入剖析

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

public class ProducerConsumerExample {

    // 定义产品类
    static class Product {
        private final int id;

        public Product(int id) {
            this.id = id;
        }

        @Override
        public String toString() {
            return "Product{" + "id=" + id + '}';
        }
    }

    // 生产者类
    static class Producer implements Runnable {
        private final BlockingQueue<Product> queue;
        private final int produceRateMillis; // 生产速率（毫秒/产品）
        private volatile boolean running = true;

        public Producer(BlockingQueue<Product> queue, int produceRateMillis) {
            this.queue = queue;
            this.produceRateMillis = produceRateMillis;
        }

        public void shutdown() {
            running = false;
        }

        @Override
        public void run() {
            int productId = 0;
            try {
                while (running) {
                    // 生产产品
                    Product product = new Product(productId++);
                    System.out.println("生产者生产了: " + product);
                    // 将产品放入队列
                    queue.put(product);
                    // 模拟生产过程耗时
                    TimeUnit.MILLISECONDS.sleep(produceRateMillis);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // 消费者类
    static class Consumer implements Runnable {
        private final BlockingQueue<Product> queue;
        private final int consumeRateMillis; // 消费速率（毫秒/产品）
        private volatile boolean running = true;

        public Consumer(BlockingQueue<Product> queue, int consumeRateMillis) {
            this.queue = queue;
            this.consumeRateMillis = consumeRateMillis;
        }

        public void shutdown() {
            running = false;
        }

        @Override
        public void run() {
            try {
                while (running) {
                    // 从队列中取出产品
                    Product product = queue.take();
                    System.out.println("消费者消费了: " + product);
                    // 模拟消费过程耗时
                    TimeUnit.MILLISECONDS.sleep(consumeRateMillis);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int queueCapacity = 10; // 队列容量
        int producerRate = 500; // 生产者每500毫秒生产一个产品
        int consumerRate = 1000; // 消费者每1000毫秒消费一个产品

        BlockingQueue<Product> queue = new LinkedBlockingQueue<>(queueCapacity);

        Producer producer = new Producer(queue, producerRate);
        Consumer consumer = new Consumer(queue, consumerRate);

        Thread producerThread = new Thread(producer);
        Thread consumerThread = new Thread(consumer);

        // 启动生产者和消费者线程
        producerThread.start();
        consumerThread.start();

        // 让生产者和消费者运行一段时间
        TimeUnit.SECONDS.sleep(10);

        // 优雅地关闭生产者和消费者
        producer.shutdown();
        consumer.shutdown();

        producerThread.join();
        consumerThread.join();

        System.out.println("生产者和消费者线程已结束。");
    }
}