【Java多线程】ArrayBlockingQueue阻塞队列原理分析（十六）

　　阅读此篇文章请先了解

　　AQS：【Java多线程】队列同步器AQS（十一）

　　阻塞队列：【Java多线程】Java中的阻塞队列（十五）

一、ArrayBlockingQueue介绍

　　ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。

　　类图结构如下：

　　

二、原理分析

2.1 示例代码　

　　设计目的：做个线程从阻塞队列中，获取元素或者放入元素，同一直接只有一个线程进行（获取或放入操作）。

　　代码

　　生产者Producer.java

public class Producer implements Runnable {
    private BlockingQueue<Integer> numbersQueue;

    public Producer(BlockingQueue<Integer> numbersQueue) {
        this.numbersQueue = numbersQueue;
    }
    public void run() {
        try {
            while(true){
                int num = ThreadLocalRandom.current().nextInt(100);
                numbersQueue.put(num);
                System.out.printf("%s：将%d号球放入队列中，存在%d个球
",Thread.currentThread().getName(), num, numbersQueue.size());
                Thread.sleep(10);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

　　消费者：Consumer.java

public class Consumer implements Runnable {

    private BlockingQueue<Integer> numbersQueue;



    public Consumer(BlockingQueue<Integer> numbersQueue) {
        this.numbersQueue = numbersQueue;
    }
    public void run() {
        try {
            while(true){

//                while (TestDemo.flag != 1) {
//                    try {
//                        Thread.sleep(1000);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }

                // 消费
                Integer num = numbersQueue.take();
                System.out.printf("%s：将%d号球从队列中取出，队列中还存在%d个球
", Thread.currentThread().getName() ,num, numbersQueue.size());

                Thread.sleep(10);

            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

　　测试TestDemo.java

public class TestDemo {

    public static volatile int flag = 0;

    public static void main(String[] args) {

        int BOUND = 10;
        int N_PRODUCERS = 10;
        int N_CONSUMERS = 10;

        BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(BOUND);

        for (int i = 0; i < N_PRODUCERS; i++) {
            new Thread(new Producer(queue), "生产者" + i).start();
        }

        for (int j = 0; j < N_CONSUMERS; j++) {
            new Thread(new Consumer(queue), "消费者" + j).start();
        }


        // 控制消费者
        // 输入不等于1 继续等待
        Scanner scanner = new Scanner(System.in);
        while (true) {
            try {
                flag = scanner.nextInt();
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

2.1 原理图　

　　1、put(e) 与 take() 流程图

　　

　　2、阻塞队列状态图