ArrayBlockingQueue 和LinkedBlockQueue

ArrayBlockingQueue

######ArrayBlockingQueue是Java多线程常用的线程安全的一个集合，基于数组实现，继承自AbstractQueue，实现了BlockingQueue和Serializable接口。 //先看看器内部的成员变量：

1private static final long serialVersionUID = -817911632652898426L;//实现了序列化接口
2
3/** 基于数组的实现，内部持有一个Object数组 */
4final Object[] items;
5
6/** 数据读取指针 */
7int takeIndex;
8
9/** 数据插入指针 */
10int putIndex;
11
12/** 当前队列中元素的总数 */
13int count;
14
15/** 采用了ReentrantLock 的实现 */
16final ReentrantLock lock;
17
18/** 标识当前队列中有可读元素 */
19private final Condition notEmpty;
20
21/** 标识当前队列可写入 */
22private final Condition notFull;
23
24//可以看到，ArrayBlockingQueue内部维护了一个takeIndex指针和一个putIndex指针，分别用于读取和写入；一个notEmpty和一个notFull，分别用于保证写入和读取的线程安全，唤醒读取和写入线程
25//再看看构造函数
26public ArrayBlockingQueue(int capacity, boolean fair) {
27    if (capacity <= 0)
28        throw new IllegalArgumentException();
29    this.items = new Object[capacity];//初始化数组
30    lock = new ReentrantLock(fair);//初始化ReentrantLock，并标识是否为公平锁
31    notEmpty = lock.newCondition();
32    notFull =  lock.newCondition();
33}
34
35//然后来看看ArrayBlockingQueue的offer方法
36
37public boolean offer(E e) {
38    checkNotNull(e);
39    final ReentrantLock lock = this.lock;
40    lock.lock();
41    try {
42        if (count == items.length)
43            //如果队列满，则添加失败。offer方法不会阻塞，put方法会阻塞
44            return false;
45        else {
46            enqueue(e);
47            return true;
48        }
49    } finally {
50        lock.unlock();
51    }
52}
53//首先做空值检查，如果为空，抛出空值异常。然后使用了ReentrantLock ，来保证offer的线程安全性。下面来看看真正的添加方法enqueue：
54private void enqueue(E x) {
55    final Object[] items = this.items;
56    items[putIndex] = x;
57    if (++putIndex == items.length)
58        putIndex = 0;
59    count++;
60    notEmpty.signal();
61}
62//可以看到，ArrayBlockingQueue内部维护了一个putIndex 指针，该指针指向当前队列可以插入的位置，直接将当前的Object对象插入到inputIndex位置，然后让inputIndex自增，如果队列已满，则指向第一个元素。最后元素总数加一，并唤醒读线程
63//最后我们来看读取take方法：
64public E take() throws InterruptedException {
65    final ReentrantLock lock = this.lock;
66    lock.lockInterruptibly();
67    try {
68        while (count == 0)
69            notEmpty.await();//take方法是阻塞的，poll方法不会阻塞，直接返回。
70        return dequeue();
71    } finally {
72        lock.unlock();
73    }
74}
75//可以看到，那么take方法将被阻塞。下面看看出对方法dequeue：
76private E dequeue() {
77    final Object[] items = this.items;
78    @SuppressWarnings("unchecked")
79    E x = (E) items[takeIndex];
80    items[takeIndex] = null;
81    if (++takeIndex == items.length)
82        takeIndex = 0;//如果取到最后一个元素，takeIndex 指向第一个元素
83    count--;//元素总数减一
84    if (itrs != null)
85        itrs.elementDequeued();
86    notFull.signal();//唤醒写入线程
87    return x;
88}

以上便是ArrayBlockingQueue的基本方法，内部锁的实现是ReentrantLock ，维护了take和put两个指针；入队和出对方法也都挺简单的，需要注意的是，take和put方法是阻塞的，offer、add、poll等方法是非阻塞的

LinkedBlockingQueue

LinkedBlockingQueue基于链表实现，继承了AbstractQueue，实现了序列化接口Serializable和BlockingQueue接口

1 //首先看看内部成员变量：
2
3private final int capacity;
4
5/** count用来记录内部元素的总数 */
6private final AtomicInteger count = new AtomicInteger();
7
8/** Node节点的头指针*/
9transient Node<E> head;
10
11/** 尾指针*/
12private transient Node<E> last;
13
14/** 读锁 */
15private final ReentrantLock takeLock = new ReentrantLock();
16
17/** Wait queue for waiting takes */
18private final Condition notEmpty = takeLock.newCondition();
19
20/** 写锁 */
21private final ReentrantLock putLock = new ReentrantLock();
22
23/** Wait queue for waiting puts */
24private final Condition notFull = putLock.newCondition();

可以看到，与ArrayBlockQueue不同，元素总数使用了原子类AtomicInteger ，内部多维护了两把锁，读锁和写锁。其实现相对更加复杂

1//下面看看其构造方法
2public LinkedBlockingQueue(int capacity) {
3    if (capacity <= 0) throw new IllegalArgumentException();//容量不能小于0
4    this.capacity = capacity;
5    last = head = new Node<E>(null);//初始化头尾指针
6}
7//下面是offer方法
8public boolean offer(E e) {
9    if (e == null) throw new NullPointerException();//不接受空值
10    final AtomicInteger count = this.count;
11    if (count.get() == capacity)//如果当前元素总数等于其容量大小，直接返回false
12        return false;
13    int c = -1;
14    Node<E> node = new Node<E>(e);
15    final ReentrantLock putLock = this.putLock;
16    putLock.lock();
17    try {
18        if (count.get() < capacity) {
19            enqueue(node);
20            c = count.getAndIncrement();
21            if (c + 1 < capacity)
22                notFull.signal();
23        }
24    } finally {
25        putLock.unlock();
26    }
27    if (c == 0)
28        signalNotEmpty();
29    return c >= 0;
30}
31//我们可以看到，LingkedBlockQueue是不接受空值的。offer是非阻塞的。入队之后，如果队列没有满，唤醒其他入队线程，并且唤醒出队线程。
32//继续看入队方法enqueue
33private void enqueue(Node<E> node) {
34    last = last.next = node;
35}//可以看到入队方法相当简单，就是把尾节点的下一个节点直接指向新加入的节点，然后将新加入的节点作为尾节点
36
37//然后看看take方法：
38public E take() throws InterruptedException {
39    E x;
40    int c = -1;
41    final AtomicInteger count = this.count;
42    final ReentrantLock takeLock = this.takeLock;
43    takeLock.lockInterruptibly();
44    try {
45        while (count.get() == 0) {
46            notEmpty.await();//take方法是阻塞的
47        }
48        x = dequeue();
49        c = count.getAndDecrement();
50        if (c > 1)
51            notEmpty.signal();
52    } finally {
53        takeLock.unlock();
54    }
55    if (c == capacity)
56        signalNotFull();
57    return x;
58}//也挺简单的，就是先判断是否可以出队，不能则等待，否则出队，然后唤醒其他出队线程，并唤醒入队线程
59//最后是出队方法：
60private E dequeue() {
61    Node<E> h = head;
62    Node<E> first = h.next;
63    h.next = h; // help GC
64    head = first;
65    E x = first.item;
66    first.item = null;
67    return x;
68} // 直接将一个元素取出，然后首位元素置空

总结，从实现来看，相比ArrayBlockQueue，LinkedBlockQueue的加锁方法相对更加复杂，但是其入队和出队方法更加简单；和ArrayBlockQueue一样，take、put方法阻塞，offer、add、poll方法不会阻塞