java并发队列之阻塞队列-ArrayBlockingQueue

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /**
     * 序列化ID。这个类依赖于默认的序列化

     *甚至对于默认序列化的项数组，即使

     *它是空的。否则，它不能被声明为.，即

     *这里需要。
     */
    private static final long serialVersionUID = -817911632652898426L;

    /** 排队的物品 */
    final Object[] items;

    /** 用于下一次获取、轮询、查看或删除的项目索引 */
    int takeIndex;

    /** 下一个put、.或add的项目索引 */
    int putIndex;

    /** 队列中的元素数量 */
    int count;

    /*
     * 并发控制使用经典的两条件算法

     *发现在任何教科书。
     */

    /** 主锁保护所有通道 */
    final ReentrantLock lock;

    /** 等待获取的条件 */
    private final Condition notEmpty;

    /** 等待放入的条件 */
    private final Condition notFull;

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

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();
    }

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();//这里并没有调用lock方法，而是调用了可被中断的lockInterruptibly，该方法可被线程中断返回，lock不能被中断返回。 
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

package com.aspire;

/**
 * 阻塞队列
 *
 * @author yanlin
 * @version v1.3
 * @date 2018-12-22 1:28 PM
 * @since v8.0
 **/

import org.testng.annotations.Test;

import java.util.concurrent.*;

public class BlockingQueueTest {
    @Test
    public void test() throws InterruptedException {
        ArrayBlockingQueue abq = new ArrayBlockingQueue(10, true);
        for (int i = 1; i <= 10; i++) {
            put(abq, i);
            get(abq);
        }
    }

    private void get(ArrayBlockingQueue abq) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        CountDownLatch countDownLatch = new CountDownLatch(2);
        for (int i = 1; i <= 2; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    Integer integer = null;
                    try {
                        integer = (Integer) abq.take();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (null == integer) {
                        System.out.println("队列是空，没有取到元素");
                    } else {
                        System.out.println("取到了一个元素是:" + integer + "---目前队列中元素个数是 :" + abq.size());
                    }
                    countDownLatch.countDown();
                }
            });
        }
        executorService.shutdown();
        countDownLatch.await();
    }

    /**
     * 添加元素
     *
     * @param abq
     * @param i
     */
    private void put(ArrayBlockingQueue abq, int i) throws InterruptedException {
        abq.put(i);
        System.out.println("存入了一个元素是   " + i);

    }
}