并发数据结构与线程（ArrayBlockingQueue）
并发数据结构与线程（ArrayBlockingQueue）
今天在QQ群上抛出来一个问题，如下
我以Java自带的数据结构为例，用源码的形式说明，如何阻塞线程、通知线程的。
一、Lock & Condition ArrayBlockingQueue以可重入锁和两个Condition对象来控制并发。
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
private final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;构造函数中初始化了notEmpty和notFull.
/**
* Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
* capacity and the specified access policy.
* @param capacity the capacity of this queue
* @param fair if <tt>true</tt> then queue accesses for threads blocked
* on insertion or removal, are processed in FIFO order;
* if <tt>false</tt> the access order is unspecified.
* @throws IllegalArgumentException if <tt>capacity</tt> is less than 1
*/
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();
}二、线程阻塞 当ArrayBlockingQueue存储的元素是0个的时候，take()方法会阻塞.
public Object take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == 0)
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to non-interrupted thread
throw ie;
}
Object x = extract();
return x;
} finally {
lock.unlock();
}
}这里take方法首先获得可重入锁lock,然后判断如果元素为空就执行notEmpty.await(); 这个时候线程挂起。
三、通知线程 比如使用put放入一个新元素，
/**
* Inserts the specified element at the tail of this queue, waiting
* for space to become available if the queue is full.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
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();
}
}在enqueue方法中，
/**
* Inserts element at current put position, advances, and signals.
* Call only when holding lock.
*/
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();
}对刚才的notEmptyCondition进行通知。
四、ReentrantLock vs AbstractQueuedSynchronizer
ArrayBlockingQueue使用ReentrantLock来控制并发，同时也使用ArrayBlockingQueue的Condition对象来与线程交互。notEmpty和notFull都是由
ReentrantLock的成员变量sync生成的，
public Condition newCondition() {
return sync.newCondition();
}sync可以认为是一个抽象类类型，Sync，它是在ReentrantLock内部定义的静态抽象类，抽象类实现了newCondition方法，
final ConditionObject newCondition() {
return new ConditionObject();
}返回的类型是实现了Condition接口的ConditionObject类，这是在AbstractQueuedSynchronizer内部定义的类。在ArrayBlockingQueue中的notEmpty就是ConditionObject实例。
阻塞：
当ArrayBlockingQueue为空时，notEmpty.await()将自己挂起，如ConditionObject的await方法，
/**
* Implements interruptible condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled or interrupted.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* </ol>
*/
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
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);
}addConditionWaiter是将当前线程作为一个node加入到ConditionObject的队列中，队列是用链表实现的。
如果是初次加入队列的情况，node.waitStatus == Node.CONDITION成立，方法isOnSyncQueue返回false，那么就将线程park。
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
....
}至此线程被挂起，LockSupport.park(this);这里this是指ConditionObject,是notEmpty.
通知：
当新的元素put进入ArrayBlockingQueue后，notEmpty.signal()通知在这上面等待的线程，如ConditionObject的signal方法，
/**
* Moves the longest-waiting thread, if one exists, from the
* wait queue for this condition to the wait queue for the
* owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}doSignal方法，
/**
* Removes and transfers nodes until hit non-cancelled one or
* null. Split out from signal in part to encourage compilers
* to inline the case of no waiters.
* @param first (non-null) the first node on condition queue
*/
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}doSignal一开始接收到的参数就是firstWaiter这个参数，在内部实现中用了do..while的形式，首先将first的的nextWaiter找出来保存到firstWaiter此时（first和firstWaiter不是一回事），在while的比较条件中可调用了transferForSignal方法，
整个while比较条件可以看着短路逻辑，如果transferForSignal结果为true,后面的first = firstWaiter就不执行了，整个while循环就结束了。
参照注释，看
transferForSignal方法，
/**
* Transfers a node from a condition queue onto sync queue.
* Returns true if successful.
* @param node the node
* @return true if successfully transferred (else the node was
* cancelled before signal)
*/
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}首先确保想要被signal的等待node还是处于Node.CONDITION状态，然后调整状态为Node.SIGNAL，这两个都是采用CAS方法，最后调用的是
LockSupport.unpark(node.thread);五、LockSupport 至此，我们已经知道了线程的挂起和通知都是使用LockSupport来完成的，并发数据结构与线程直接的交互最终也是需要LockSupport。那么关于LockSupport，我们又可以了解多少呢？
Ref: Java中的ReentrantLock和synchronized两种锁定机制的对比 Java的LockSupport.park()实现分析