阻塞队列ArrayBlockingQueue的实现原理浅析

阻塞队列介绍

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

阻塞队列例子

    public static void main(String[] args) {
        BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    int i = 0;
                    while(true) {
                        i++;
                        System.out.println("放入："+i);
                        bq.put(i+""+new Date());
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while(true) {
                        String i =bq.take();
                        System.out.println("取出："+i);
                        Thread.sleep(2000);
                    }
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

阻塞队列原理

我们这里简要分析一下上面的程序是如何实现一个线程（生产者）放内容，等队列满了就暂停等待，直到队列有空位就继续放，另一个线程（消费者）取内容，队列为空就暂停等待，知道到队列有值就继续取。

1、初始化阻塞队列

BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);

我们这里是初始化长度为10的阻塞队列，点进去看看具体做了啥

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
       throw new IllegalArgumentException();
    //1、初始化一个长度为capacity的数组，这里传的是10
    this.items = new Object[capacity];
    //2、初始化一个可重入锁
    lock = new ReentrantLock(fair);
    //3、初始化一个非空Condition
    notEmpty = lock.newCondition();
    //4、初始化一个非满Condition
    notFull =  lock.newCondition();
}

上面代码简单清晰易懂，第1步初始化一个长度为capacity的数组是指定这个阻塞队列的长度就是10，第2步是初始化一个可重入锁，这个是用于在队列中放数据的时候和取数据的时候做同步的，应为不可能说生产者和消费者同时放数据和取数据，会有并发问题，所以这里需要一个锁来同步，同步生产者和消费者在放数据和取数据的交互过程，生产者和消费者在跟队列放或者取完数据后就可以把锁释放去执行自己的业务逻辑了。

第3步和第4步是生成了两个Condition，那Condition是干啥的呢，这里简单说下，在使用Lock之前，我们都使用Object 的wait和notify实现同步的。如下代码

synchronize(obj){ 
    obj.wait();//等待
}

synchronize(obj){ 
    obj.notify();//唤醒 
}

有了lock后，现在是：

lock.lock(); 
condition.await(); //等待
lock.unlock();

lock.lock(); 
condition.signal(); //唤醒
lock.unlock();

相比Object更加的灵活。所以可以推测第3步和第4步就是用来在队列空或者满的时候进行阻塞和唤醒的，具体怎么用，我们得继续向下看源码。

2、生产者放数据

bq.put(i+""+new Date());

点进去查看源码

public void put(E e) throws InterruptedException {
        //1、检查元素是否是空的
        checkNotNull(e);
        //2、获取重入锁
        final ReentrantLock lock = this.lock;
        //3、锁住
        lock.lockInterruptibly();
        try {
            //4、如果队列满了
            while (count == items.length)
                //await阻塞当前线程，释放锁
                notFull.await();
            //5、如果没有满，则加入到队列中
            enqueue(e);
        } finally {
            //不管如何，最后都需要进行锁的释放
            lock.unlock();
        }
    }

逻辑还是特别清晰的，关键点就在于如果队列满了，就直接阻塞线程，也就是阻塞了生产者。

 notFull.await();

所以初始化的时候实例化的notFull的作用就是给生产者用的，可以推测，在消费者取走数据后，肯定会调用唤醒的方法。

notFull.signal();

也可以推测出，如果这里不满，那么执行完 enqueue(e)加入数据到队列中后，肯定会调用消费者线程的唤醒方法。

notEmpty.signal();

我们进入enqueue(e)方法的源码看看是否如推测一样。

private void enqueue(E x) {
       // assert lock.getHoldCount() == 1;
       // assert items[putIndex] == null;
       //获取队列
       final Object[] items = this.items;
       //新加入一个值
       items[putIndex] = x;
       //如果已经加入到最后了，那么下一次从第一位加入，否则在下一个位置加入
       if (++putIndex == items.length)
           putIndex = 0;
       //队列中的数目自增
       count++;
       //唤醒消费者
       notEmpty.signal();
    }

逻辑也很简单，不出所料，最后因为在队列中加入了数据，调用了消费在的Condition进行唤醒。

notEmpty.signal();

接下来我们再看看消费者怎么消费数据的，如果队列空了会怎么阻塞，其实都可以菜猜到，肯定是先获取锁，然后判断队列是否为空，若为空则阻塞，不为空则从队列中取树，后唤醒生产者，如果生产者本来就没有被阻塞也就不用唤醒。

3、消费者消费数据

String i =bq.take();

我们点进源码去看看

 public E take() throws InterruptedException {
        //1、获取锁
        final ReentrantLock lock = this.lock;
        //2、锁住
        lock.lockInterruptibly();
        try {
            //3、如果队列为空，则阻塞消费者
            while (count == 0)
                //4、阻塞，这里会释放锁
                notEmpty.await();
            //5、如果不为空，这里会取数据
            return dequeue();
        } finally {
            //7、释放锁
            lock.unlock();
        }
    }

逻辑也很简单，反正一开始只要生成则没有在放数据的过程中，或者已经阻塞，则消费者在第2步骤都可以获得锁进行操作，我们直接看如果不为空dequeue()如何取数据，可以推测，里面取完数据后会进行唤醒生产者的操作，点进去看看。

 private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        //获取当前数组
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        //取的数据
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        //下一次就取下一个，但是如果已经到最后了，下一次就取第一个
        if (++takeIndex == items.length)
            takeIndex = 0;
        //队列数据减一
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        //唤醒生产者
        notFull.signal();
        return x;
    }

源码也很简单，逻辑清楚，最后也不出所料调用了唤醒生产者的方法。

notFull.signal();

总结

这大概是我见过JUC中最清晰易懂的源码了。