深入探索Java并发编程：ArrayBlockingQueue详解

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ProducerConsumerWithArrayBlockingQueue {

    public static void main(String[] args) {
        // 创建一个容量为10的ArrayBlockingQueue作为生产者和消费者的共享队列
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
        
        // 创建一个容量为5的ArrayBlockingQueue用于存储消费者的处理结果
        BlockingQueue<Integer> resultQueue = new ArrayBlockingQueue<>(5);
        
        // 创建一个AtomicInteger作为数据生成的计数器
        AtomicInteger counter = new AtomicInteger();
        
        // 创建一个生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 20; i++) {
                    int item = counter.incrementAndGet();
                    System.out.println("生产者生产数据：" + item);
                    // 将数据放入队列中
                    queue.put(item);
                    // 稍微延迟一下，模拟生产数据的时间消耗
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 创建一个固定线程池的ExecutorService用于执行消费者任务
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        
        // 提交3个消费者任务到线程池中
        for (int i = 0; i < 3; i++) {
            executorService.submit(() -> {
                try {
                    while (true) {
                        // 从队列中取出数据
                        int item = queue.take();
                        // 处理数据（此处仅打印作为示例）
                        System.out.println("消费者" + Thread.currentThread().getId() + "消费数据：" + item);
                        // 假设处理后的数据是原始数据的平方
                        int processedItem = item * item;
                        // 将处理后的结果存入结果队列中
                        resultQueue.put(processedItem);
                        // 稍微延迟一下，模拟处理数据的时间消耗
                        Thread.sleep(500);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        // 启动生产者线程
        producer.start();
        
        // 等待生产者线程完成
        try {
            producer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        // 关闭ExecutorService（这将导致消费者线程中断）
        executorService.shutdown();
        try {
            // 等待一段时间，让消费者线程处理剩余的数据
            if (!executorService.awaitTermination(2, TimeUnit.SECONDS)) {
                executorService.shutdownNow(); // 如果超时则强制关闭消费者线程
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        // 处理结果队列中的数据（此处仅打印作为示例）
        while (!resultQueue.isEmpty()) {
            System.out.println("处理结果：" + resultQueue.poll());
        }
    }
}