阻塞队列（BlockingQueue）

import java.util.Random;
import java.util.concurrent.*;

public class BlockingQueueTest1 {

    /**
     * 生产者
     */
    public static class Producer implements Runnable {

        /**
         * 阻塞队列
         */
        private BlockingQueue<Integer> blockingQueue;

        /**
         * 判断是否循环
         */
        private boolean isRunning = true;

        /**
         * 随机数据范围
         */
        private static final int RANGE_FOR_DATA = 1000;

        private Random random = new Random();

        public Producer(BlockingQueue<Integer> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }

        @Override
        public void run() {
            while (isRunning) {
                try {
                    /** 生产出随机数 */
                    int data = random.nextInt(RANGE_FOR_DATA);
                    System.out.println(Thread.currentThread().getName() + " 生产数据:" + data);
                    /** 将随机数放入阻塞队列 */
                    blockingQueue.put(data);
                    System.out.println(Thread.currentThread().getName() + " 插入队列:" + data);
                    /** 进行随机时间休眠 */
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
                    System.out.println("程序结束啦，我不用再等待阻塞队列有空余位置了！");
                }
            }
        }

        /**
         * 终止生产线程
         */
        public void shutDown() {
            isRunning = false;
        }
    }

    /**
     * 消费者
     */
    public static class Consumer implements Runnable {

        /**
         * 阻塞队列
         */
        private BlockingQueue<Integer> blockingQueue;

        /**
         * 判断是否循环
         */
        private boolean isRunning = true;

        /**
         * 随机数据范围
         */
        private Random random = new Random();

        public Consumer(BlockingQueue<Integer> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }

        @Override
        public void run() {
            while (isRunning) {
                try {
                    /** 从阻塞队列中获取随机数 */
                    int data = (int) blockingQueue.take();
                    System.out.println(Thread.currentThread().getName() + " 消费数据:" + data);
                    /** 进行随机时间休眠 */
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
                    System.out.println("程序结束啦，我不用再等待阻塞队列非空了！");
                }
            }
        }

        /**
         * 终止消费线程
         */
        public void shutDown() {
            isRunning = false;
        }
    }

    public static void main(String[] args) {
        /** 创建容量大小为5的阻塞队列 */
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>(5);
        /** 创建连接池 */
        ExecutorService pool = Executors.newCachedThreadPool();
        /** 创建生产线程，消费线程各5个 */
        Producer[] producers = new Producer[5];
        Consumer[] consumers = new Consumer[5];
        /** 实例化生产线程与消费线程并且执行线程 */
        for (int i = 0; i < producers.length; i++) {
            producers[i] = new Producer(blockingQueue);
            consumers[i] = new Consumer(blockingQueue);
            pool.execute(producers[i]);
            pool.execute(consumers[i]);
        }
        try {
            /** 等待5秒后进行手动中断 */
            Thread.sleep(5 * 1000);
            for (int i = 0; i < producers.length; i++) {
                producers[i].shutDown();
                consumers[i].shutDown();
            }
            /** 其实提不提醒线程关闭都一个样了，阻塞的线程，不会因为手动中断而中断的 */
            pool.shutdown();
            /** 等待2秒，若还有线程没有关闭则强行中断所有等待线程 */
            if (!pool.awaitTermination(2 * 1000, TimeUnit.MILLISECONDS)) {
                /** 超时的时候向线程池中所有的线程发出中断 */
                pool.shutdownNow();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}