ArrayBlockinQueue及Condition源码解析

不念不忘少年蓝@ 2022-05-09 13:04 180阅读 0赞

在进入ArrayBlockingQueue源码之前，我们先看看AbstractQueuedSynchronizer类的内部结构：  
![AbstractQueuedSynchronizer内部结构][AbstractQueuedSynchronizer]  
其有一个重要的静态内部类Node，将一个线程包装成一个Node，每个Node会持有prev，next，nextWaiter 节点的引用。

AbstractQueuedSynchronizer里的state属性就表示被加锁的次数，如果state>0，则表示当前锁对象被其他线程持有。  
AbstractQueuedSynchronizer和其内部类ConditionObject都维护了一条Node的链表，且都持有链表的头尾节点，AbstractQueuedSynchronizer通过prev，next 来实现双向的链表；而ConditionObject借助nextWaiter实现单向的链表。

ArrayBlockingQueue在实例化是会调用如下构造函数：

final ReentrantLock lock;
        private final Condition notEmpty;
        private final Condition notFull;
        
    	public ArrayBlockingQueue(int capacity, boolean fair) { 
    	    if (capacity <= 0)
    	        throw new IllegalArgumentException();
    	    this.items = new Object[capacity];
    	    lock = new ReentrantLock(fair);
    	    notEmpty = lock.newCondition();
    	    notFull =  lock.newCondition();
    	}

默认情况下使用可重入的排他的非公平锁，同时通过lock创建两个Condition对象，notEmpty和notFull共用同一个锁。

接下来我们看看ArrayBlockingQueue的两个阻塞方法，put(Object)和take() :

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();
            }
        }
    
    	public E take() throws InterruptedException { 
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
                while (count == 0)
                    notEmpty.await();
                return dequeue();
            } finally { 
                lock.unlock();
            }
        }

put和take方法都在lock的包围之中，也就是说同一时间只能有一个线程能够执行put或take方法，向容器内加入或者取出元素。

如果某个线程执行put后者take方法，抢到了锁，那么会将AQS的state属性通过CAS增加1，然后进入try代码块。

在高并发下，同一时间肯定有大量线程线下执行put或者take方法，但只有一个能够获取到锁，其他的线程都会被阻塞。对于未阻塞直接抢到锁的线程，其不会加入到Node链表中；对于未抢到锁而被阻塞的线程，会被封装为Node对象，添加到AQS的成员变量中，以链表的形式存储在head和tail之间。

![阻塞队列方法链表图][70]

每个被阻塞的线程封装成一个Node，通过CAS设置为链表的新tail，其prev节点为之前的tail，他们在链表上的顺序就代表着他们抢锁失败的顺序。

在最开始时，第一个线程没有抢到锁，先创建一个空的Node，赋值给head，然后将此线程封装为一个Node，赋值给tail，维护head和tail之间的引用，之后没有抢到锁的线程按顺序依次向链表末尾添加，赋值为tail。

当第一个直接抢到锁的线程执行unlock方法时，其会唤醒head节点的next，也就是上图中的put( )方法所在线程，此时这个线程被唤醒，重新执行抢锁操作，有如下两种可能结果：

1.  抢锁失败：因为是非公平锁，中途有一个额外的线程插进来，抢到了锁，head的next节点所表示的线程抢锁再次失败，被再次阻塞，等待新插进来的线程执行unlock来再次唤醒，重复上述流程
2.  抢锁成功：head的next节点将自身设置为新的head节点，之后其执行unlock方法时，也唤醒自身的next节点，重复上述流程

可重入的非公平锁就是按照上述流程来实现线程的互斥的，当加入了Condition之后，又有了一些额外的变化。

当顺利的抢到锁，进入try代码块之后，可能因为某些原因导致不满足后续执行条件，因而执行condition的await方法。在await方法中，首先会将当前线程再次封装成一个Node节点，添加到Condition对象的属性中，如果firstWaiter为null，那么赋值给firstWaiter；如果firstWaiter有值，那么将此Node定义为新的lastWaiter，之前的lastWaiter.next = Node。

![在这里插入图片描述][70 1]

多个线程调用同一个Condition对象的await方法，会按照调用顺序依次添加到Condition对象的Node节点链表中，ArrayBlockingQueue里的lock生成了两个Condition，每个Condition存储对应方法的Node节点链表，一个put，一个take。

在添加到waiter链表中后，会释放当前的锁，也就是将AQS的state属性置为0，然后唤醒head节点的next节点代表的线程。之后会进入一个while循环，当await方法生成的Node节点没有被加入到head，tail所代表的链表中，那么会阻塞当前线程。正常情况下，执行await方法后当前线程会被阻塞，直到其对应的在waiter链表中的Node节点被添加到head，tail的链表中。

当前线程通过await方法阻塞了，那么就需要其他线程执行同一个Condition的signal方法来唤醒当前线程。比如put方法被阻塞，那么需要其他线程调用put对应Condition的signal。在await方法中会释放持有的锁，同时唤醒head节点的next对应的线程，一般来说此线程最终会调用await对应Condition的signal方法。

public class ConditionObject implements Condition, java.io.Serializable { 
    
    	public final void await() throws InterruptedException { 
                if (Thread.interrupted())
                    throw new InterruptedException();
                // 将当前线程封装为Node，添加到waiter链表中
                Node node = addConditionWaiter();
                // 释放锁，唤醒AQS内head的next节点所代表的线程
                int savedState = fullyRelease(node);
                int interruptMode = 0;
                // 如果addConditionWaiter()方法生成的Node不在head,tail链表中,正常情况下是的
                while (!isOnSyncQueue(node)) { 
                	// 阻塞当前线程,直至addConditionWaiter()生成的Node成为了firstWaiter,且被signal()唤醒
                    LockSupport.park(this);
                    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                        break;
                }
                if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                    interruptMode = REINTERRUPT;
                if (node.nextWaiter != null) // clean up if cancelled
                    unlinkCancelledWaiters();
                if (interruptMode != 0)
                    reportInterruptAfterWait(interruptMode);
            }
    
    }

在await方法中，通过当前线程生成另外一个Node，被加入到waiter链表中，此时当前线程对应的Node可能有以下两种情形：

1.  当前线程是中途插队抢到锁的，所以未被阻塞过，也就是说其不存在head，tail的链表中。当前线程仅仅对应一个Node，在waiter链表上。
2.  当前线程是阻塞唤醒后抢到锁的，那么在执行当前线程时，其对应的Node应该就是head节点。在await方法内唤醒head的next节点，也就是此节点的next对应的线程。如果next线程抢到锁了，那么会将next设置为新的head；如果没有抢到锁，也就是被中途插入的线程抢到锁了，在中途插入的线程释放锁后，再次唤醒head的next线程。

综上两种情形，**当await方法执行时，当前线程重新生成了一个Node，转移到了waiter链表上。可能还有另一个Node在head，tail链表上，是head，但是已经执行过了，不会再次被唤醒**

在Condition的signal方法中，其会将firstWaiter节点的Node添加到head，tail链表中，赋值为新的tail，然后唤醒Node节点对应的线程，同时将firstWaiter指向之前first的nextWaiter，也就是下一个Node。

public class ConditionObject implements Condition, java.io.Serializable { 
    
    	public final void signal() { 
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                Node first = firstWaiter;
                if (first != null)
                	// 唤醒firstWaiter所代表的线程
                    doSignal(first);
        }
    
    	private void doSignal(Node first) { 
                do { 
                	// 将firstWaiter 指向 其 nextWaiter, 也就是向后移动一位
                    if ( (firstWaiter = first.nextWaiter) == null)
                        lastWaiter = null;
                    first.nextWaiter = null;
                } while (!transferForSignal(first) &&
                         (first = firstWaiter) != null);
            }
    
    	final boolean transferForSignal(Node node) { 
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
            // 将firstWaiter指向的Node添加到head,tail的链表中,设置为新的tail节点
            Node p = enq(node);
            int ws = p.waitStatus;
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            	// 唤醒firstWaiter关联的线程,也就是上文中await()方法阻塞的一个线程
                LockSupport.unpark(node.thread);
            return true;
        }
    }

**在signal方法中，将await方法生成的Node从waiter链表上重新转移至head，tail链表上设置为新的tail**

在await( ) 方法内 被阻塞的线程被唤醒后，在while循环内，再次判断当前Node是否在head，tail链表中，会返回true，也就是能够跳出while循环：

// 判断await生成的Node是否被加入到head,tail的链表中
    		while (!isOnSyncQueue(node)) { 
             	// 阻塞当前线程,直至addConditionWaiter()生成的Node成为了firstWaiter,且被signal()唤醒
                 LockSupport.park(this);
                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                     break;
             }

接下去执行acquireQueued（）方法， 尝试获得锁：

final boolean acquireQueued(final Node node, int arg) { 
            boolean failed = true;
            try { 
                boolean interrupted = false;
                for (;;) { 
                	// 获取当前Node的prev,也就是前一个节点
                    final Node p = node.predecessor();
                    // 如果前一个节点是head,那么当前线程尝试去获取锁
                    if (p == head && tryAcquire(arg)) { 
                    	// 如果获取成功了，那么将当前节点设置为新的head,结束流程
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    // 如果不满足上述条件，那么阻塞当前线程,直至被其他线程用signal再次唤醒
                    if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }

如果await代表的线程被重新唤醒后，有下列两种情形：

1.  如果其位置不是head的next节点，那么会被重新阻塞，直至其位置前移到head的next，被head的unlock唤醒
2.  如果其位置是head的next，那么其会重新尝试去抢锁，如果没抢到，那么阻塞，等待其他线程的unlock唤醒；如果抢到了，那么设置自设为新的head，继续执行。

如果await的Node位置是head的next，那么signal会唤醒此Node对应的线程，而signal( ) 外围lock.unlock( ) 方法也能唤醒此Node，也就是一个线程的signal和unlock两次唤醒head的next节点，这中间存在着多种可能性，感兴趣的可以自己思考下，不过因为同一时间 只有一个线程能够获得锁，基本不会出现线程安全问题，这里不再讨论。

await，signal的简要流程图如下：

![await，signal的流程图][await_signal]

以上就是ArrayBlockingQueue里阻塞方法put和take的源码流程了，写的不是很好，逻辑不是很清晰，如果有什么问题可以在博客下留言，我会尽量回复。

下面做一个总结：

1.  一个lock可以new多个Condition，也就是上图中会有多个waiter链表，每个Condition的await和signal都只操作自身的waiter链表节点，但他们共用同一个Lock的head，tail链表
2.  一般只有head节点Thread在运行，其他节点都被阻塞，也就是await和signal运行的线程都是head节点所在的线程
3.  await方法会将head的Thread封装成另一个Node，追加到waiter链表的末尾，然后释放锁，唤醒head的next节点。流程就相当于将head节点转移到waiter链表的lastWaiter，唤醒head的next
4.  signal方法会将firstWaiter节点转移到head，tail链表，追加到末尾
5.  lock.unlock（）方法会释放锁，然后唤醒head的next节点所在的线程

[AbstractQueuedSynchronizer]: /images/20220506/b8e49db9e78e4a2e9b09acedca732ac3.png
[70]: /images/20220506/52814f8ec379473fa9395802efb477aa.png
[70 1]: /images/20220506/07f7255d917247a9b6465ac21776a6b7.png
[await_signal]: /images/20220506/3ecea762597343ffadcda4d4bf35c5e0.png