Java并发编程实战(学习笔记 十三 第十四章 构建自定义的同步工具 下 )

14.3 显式的Condition对象

13章中介绍，在某些情况下，当内置锁过于灵活时，可以使用显式锁。

正如Lock时一种广义的内置锁，Condition时一种广义的内置条件队列

//       14-10     Condition接口
public interface Condition {
void await() throws InterruptedException;
boolean await(long time, TimeUnit unit)
throws InterruptedException;
long awaitNanos(long nanosTimeout) throws InterruptedException;
void awaitUninterruptibly();
boolean awaitUntil(Date deadline) throws InterruptedException;
void signal();
void signalAll();
}

内置条件队列存在一些缺陷，每个内置锁都只能有一个相关联的条件队列，因而在像BoundedBuffer这种类中，多个线程可能在同一个条件队列上等待不同的条件谓词，并在最常见的加锁模式下公开条件队列对象。

如果想编写一个带有多个条件谓词的并发对象，或者想获得除了条件队列可见性的更多控制权，就可以使用显式的Lock和Condition而不是内置锁和条件队列。

一个Condition和一个Lock关联在一起，就像一个条件队列和一个内置锁相关联一样。要创建一个Condition，可以在相关联的Lock上调用Lock.newCondition方法。

Condition比内置条件队列提供了更丰富的功能：在每个锁上可存在多个等待，条件等待可以时可中断的或不可中断的，基于时限的等待，以及公平的非公平的队列操作。

与内置条件队列不同的时，对于每个Lock，可以有任意数量的Condition对象。Condition对象继承了相关的Lock对象的公平性，对于公平的锁，线程会依照FIFO顺序从Condition.await中释放。

在Condition对象，与wait,notify，notifyAll方法分别对应的时await，signal，signalAll。但是Condition对Object进行了扩展，因而它也包含wait和notify方法，一定要确保使用正确的版本——await和signal。

14-11使用两个Condition，分别为notFull和notEmpty，用于表示”非空“和”非满“两个条件谓词。当缓存为空时，take将阻塞并等待notEmpty，此时put想notEmpty发送信号，可以解除任何在take中阻塞的线程。

//     14-11  使用显式条件变量的有界缓存
@ThreadSafe
public class ConditionBoundedBuffer <T> {
protected final Lock lock = new ReentrantLock();
//条件谓词：notFull(count<items.length)
private final Condition notFull = lock.newCondition();
//条件谓词：noEmpty(count>0)
private final Condition notEmpty = lock.newCondition();
private static final int BUFFER_SIZE = 100;
@GuardedBy("lock") private final T[] items = (T[]) new Object[BUFFER_SIZE];
@GuardedBy("lock") private int tail, head, count;

//阻塞直到notFull
public void put(T x) throws InterruptedException {
lock.lock();
try {
while (count == items.length)
notFull.await();
items[tail] = x;
if (++tail == items.length)
tail = 0;
++count;
notEmpty.signal();//put想notEmpty发送信号，可以解除任何在take中阻塞的线程。
} finally {
lock.unlock();
}
}

//阻塞直到notEmpty
public T take() throws InterruptedException {
lock.lock();
try {
while (count == 0)
notEmpty.await();
T x = items[head];
items[head] = null;
if (++head == items.length)
head = 0;
--count;
notFull.signal();
return x;
} finally {
lock.unlock();
}
}
}

在分析使用多个Condition的类时，比分析一个使用单一内部队列加多个条件谓词的类简单得多。通过将两个条件谓词分开并放到两个等待线程集中，Condition使其更容易满足单次通知的需求。

signal比signalAll更高效，它嫩极大地减少在每次缓存操作中发生的上下文切换与锁请求的次数。

当使用显式的Lock和Condition时，也必须满足锁，条件谓词和条件变量之间的三元关系。

如果需要一些高级功能，例如使用公平的队列操作或者在每个锁上对应多个等待线程集，那么应该优先使用Condition而不是内置条件队列。

14.4 Synchronizer（同步器）分析

ReentrantLock和Semaphore这两个接口之间存在许多共同点，这两个类都已用做一个“阀门”，即每次只允许一定数量的线程通过，并当线程到达阀门时，可以通过（在调用lock或acquire时成功返回），也可以等待（在调用lock或acquire时阻塞），还可以取消（在调用tryLock或tryAuquire时返回”假”，表示在指定的时间内锁时不可用的或无法获得许可）。

而且，这两个接口都支持可中断的，不可中断的以及限时的获取操作，也支持等待线程执行公平或非公平的队列操作。

我们可以用锁来实现计数信号量，以及可以通过计数信号量来实现锁

这并非java.util.concurrent.Semaphore的真实现方式

//         14-12  使用Lock来实现信号量
@ThreadSafe
public class SemaphoreOnLock {
private final Lock lock=new ReentrantLock();
//条件谓词：permitsAvailable(permits>0)
private final Condition permitsAvailable=lock.
4000
newCondition();
private int permits;

SemaphoreOnLock(int initialPermits) {
lock.lock();
try{
permits=initialPermits;
}finally {
lock.unlock();
}
}
//阻塞直到:permitsAvailable
public void acquire() throws InterruptedException{
lock.lock();
try{
while(permits<=0)
permitsAvailable.await();
--permits;
}finally{
lock.unlock();
}
}
public void release(){
lock.lock();
try{
++permits;
permitsAvailable.signal();
}finally {
lock.unlock();
}
}
}

事实上，ReentrantLock和Semaphore都使用了一个共同的基类，即AbstractQueueSynchronizer(AQS)，这个类也是其他许多同步类的基类。

AQS是一个用于构建锁和同步器的框架，许多同步器都可以通过AQS很容易并且高效地构造出来。

AQS解决了在实现同步器时设计的大量细节问题，例如等待线程采用FIFO队列操作顺序。在不同的同步器还可以定义一些灵活的标准来判断某个线程是应该通过还是等待。

基于AQS来构建同步器能带来许多好处。它不仅能极大地减少实现工作，而且也不必处理在多个位置上的竞争问题。

14.5 AbstractQueueSynchronizer

多数情况下不会直接使用AQS，标准同步器类集合能满足绝大多数情况的需求。

在基于AQS构建的同步器类中，最基本的操作包括各种形式的获取操作和释放操作。

获取操作是一种依赖状态的操作，并且通常会阻塞。

在使用锁或信号量时，获取(acquire)操作获得的是锁或许可，并且调用者可能会一直等待直到同步器类处于可被获取的状态。

在使用CountDownLatch时，获取(countDown)操作意味这“等待并直到闭锁到达结束状态”

使用FutureTask时，获取(get)操作意味着“等待并直到任务已经完成”。

释放并不是一个可阻塞的操作，当执行释放操作时，所有在请求时被阻塞的线程都会开始执行。

如果一个类想成为状态依赖的类，必须拥有一些状态。

AQS负责管理同步器类中的状态，它管理了一个整数状态信息，可以通过getState，setState以及compareAndSetState等protected类型方法来进行操作。

ReentrantLock用它来表示所有者线程已经重复获取该锁的次数。

Semaphore用它表示剩余的许可数量（acquire和release操作）。

FutureTask用它来表示任务的状态（尚未开始，正在运行，以完成以及已取消）。

还可以自行管理一些额外的状态变量，例如，ReentrantLock保存了锁的当前所有者的信息，这样就能区分某个操作是重入还是竞争的。

下面给出了AQS中的获取操作与释放操作的形式。

根据同步器的不同，获取操作可以是一种独占操作（例如ReentrantLock）也可以是一个非独占操作（例如Semaphore和CountDownLatch）。

//        14-13   AQS中获取操作与释放操作的标准形式
boolean acquire() throws InterruptedException {
while (当前状态不允许获取操作) {
if (需要阻塞获取请求) {
如果当前线程不在队列中，则将其插入队列
阻塞当前线程
}
else
返回失败
}
可能更新同步器的状态
如果线程处于队列中，则将其移出队列
返回成功
}
void release() {
更新同步器的状态
if (新的状态允许某个被阻塞的线程获取成功)
解除队列中一个或多个线程阻塞状态
}

如果某个同步器支持独占的获取操作，那么需要实现一些保护方法，包括tryAcquire，tryRelease和isHeldExclusively等

对于支持共享获取的同步器，则应该实现tryAuquireShared和tryReleaseShared。

AQS中的acquire, acquireShared, release和releaseShared等方法都将调用这些方法在子类中带有前缀try的版本来判断某个操作是否能执行。

在同步器的子类中，可以根据其获取操作和释放操作的语义，使用getState，setState以及compareAndSetState来检查和更新状态，并通过返回的状态值来告知基类“获取”或“释放”同步器的操作是否成功。

14-14是一个使用AQS实现的二元闭锁。它包含两个公有方法：await和signal，分别对应获取操作和释放操作。

起初，闭锁是关闭的，任何调用await的线程都将阻塞并直到闭锁被打开。当通过调用signal打开闭锁时，所有等待中的线程都将被释放，并且随后到达闭锁的线程也被允许执行。

//      14-14   使用AbstractQueueSynchronizer实现的二元锁
@ThreadSafe
public class OneShotLatch {
private final Sync sync=new Sync();
//AQS中的acquire, acquireShared, release和releaseShared等方法都将调用这些方法在子类中带有前缀try的版本来判断某个操作是否能执行。
public void signal(){
sync.releaseShared(0);
}
//起初，闭锁是关闭的(0)，任何调用await的线程都将阻塞并直到闭锁被打开
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(0);
}

private class Sync extends AbstractQueuedSynchronizer{
protected int tryAcquireShared(int ignored){
//如果闭锁时打开的(state==1)，那么这个操作将成功，否则将失败
return (getState()==1)?1:-1;
}
//在tryReleaseShared中将闭锁状态设置为打开，（通过返回值）表示该同步器处于完全被释放的状态，因而AQS让所有等待中的线程都重新尝试请求该同步器，并且由于tryReleaseShared将返回成功，因此现在的请求操作将成功
protected boolean tryReleaseShared(int ignored){
setState(1);  //现在打开闭锁
return true;   //现在其他的线程可以获取该闭锁
}
}
}

在OneShotLatch中，AQS状态用来表示闭锁状态——关闭（0）或打开（1）。await方法调用AQS的acquireSharedInterruptibly，然后接着调用OneShotLatch中的tryAcquireShared方法。

在tryAcquireShared的实现中必须返回一个值来表示这个该获取操作能否执行。

如果之前已经打开了闭锁，那么tryAcquireShared将返回成功并允许线程通过，否则就会返回一个表示获取操作失败的值。

acquireSharedInterruptibly处理失败的方式，是把这个线程放入等待队列中。

类似地，signal将调用releaseShared，接下来又会调用tryReleaseShared。

在tryReleaseShared中将闭锁状态设置为打开，（通过返回值）表示该同步器处于完全被释放的状态，因而AQS让所有等待中的线程都重新尝试请求该同步器，并且由于tryReleaseShared将返回成功，因此现在的请求操作将成功。

14.6 java.util.concurrent同步器类中的AQS

java.util.concurrent中的许多阻塞类，例如ReentrantLock，Semaphore，ReentrantReadWriteLock，CountDownLatch，SynchronousQueue和FutureTask等，都是基于AQS构建的。

14.6.1 ReentrantLock

ReentrantLock只支持独占方式的获取操作，因此它实现了tryAcquire，tryRelease和isHeldExclusively，14-15给出了非公平版本的tryAcquire。ReentrantLock将同步状态用于保存锁获取操作的次数，并且还维护一个owner变量来保存当前所有者线程的标识符，只有在当前线程刚刚获取到锁，或者正要释放锁额时候，才会修改这个变量。

在tryRelease中检查owner域，从而确保当前线程正在执行unlock操作之前已经获取了锁：在tryAcquire中将使用这个域来区分获取操作时重入的还是竞争的。

//      14-15  基于非公平的ReentrantLock实现tryAcquire
protected boolean tryAcquire(int ignored) {
final Thread current = Thread.currentThread();
int c = getState();
//如果锁未被持有，那么它将尝试更新锁的状态以表示锁已经被持有。
if (c == 0) {
//使用compareAndSetState来原子地更新状态，表示这个锁已经被占有，并确保状态在最后一次检查以后就没有被修改过
if (compareAndSetState(0, 1)) {
owner = current;
return true;    //（通过返回值）表示该同步器处于完全被释放的状态
}
} else if (current == owner) {//如果锁状态表明它已经被持有，并且如果当前线程是锁的拥有者，那么获取计数会递增，如果当前线程不是锁的拥有者，那么获取操作将失败
setState(c+1);
return true;
}
return false;
}

在一个线程尝试获取锁时，tryAcquire将首先检查锁的状态。

如果锁未被持有，那么它将尝试更新锁的状态以表示锁已经被持有。

由于状态可能在检查后被立即修改，因此tryAcquire使用compareAndSetState来原子地更新状态，表示这个锁已经被占有，并确保状态在最后一次检查以后就没有被修改过。

根据15.2可以知道,如果c的值在这个过程中没有被修改，仍为0，则变为1，表示这个锁已经被占有，并返回true，否则返回false。这是为了避免状态在检查后立即被修改。

如果锁状态表明它已经被持有，并且如果当前线程是锁的拥有者，那么获取计数会递增，如果当前线程不是锁的拥有者，那么获取操作将失败。

ReentrantLock还利用了AQS对多个条件变量和多个等待线程集的内置支持。Lock.newCondition将返回一个新的ConditionObject实例，这是AQS的一个内部类。

14.6.2 Semaphore与CountDownLatch

Semaphore将AQS的同步状态用于保存当前可用许可的数量。

14-16中的tryAcquireShared方法首先计算剩余许可的数量，如果没有足够的许可，那么会返回一个值表示获取操作的失败。

如果还有剩余的许可，那么tryAcquireShared会通过compareAndSetState以原子方式来降低许可的计数。

如果这个操作成功（意味着许可的计数自从上一次读取后就没有被修改过）那么将返回一个值表示获取操作成功。在返回值中还包含了表示其他共享获取操作能否成功的信息，如果成功，那么其他等待的线程同样会解除阻塞。

CAS(compareAndSet)算法的过程是这样：它包含3个参数CAS(V,E,N)。V表示要更新的变量(内存值)，E表示预期值(旧的)，N表示新值。当且仅当V值等于E值时，才会将V的值设为N，如果V值和E值不同，则说明已经有其他线程做了更新，则当前线程什么都不做。最后，CAS返回当前V的真实值。

//    14-16  Semaphore中的tryAuquireShared与tryReleaseShared
protected int tryAcquireShared(int acquires) {
while (true) {
int available = getState();
int remaining = available - acquires; //首先计算剩余许可的数量
//如果还有剩余的许可，那么tryAcquireShared会通过compareAndSetState以原子方式来降低许可的计数。
if (remaining < 0
|| compareAndSetState(available, remaining))
//当没有足够的许可，或者当tryAuquireShared可以通过原子方式来更新许可的计数以响应获取操作时，while循环将终止。
return remaining;
}
}
protected boolean tryReleaseShared(int releases) {
while (true) {
int p = getState();
if (compareAndSetState(p, p + releases))
return true;
}
}

当没有足够的许可，或者当tryAuquireShared可以通过原子方式来更新许可的计数以响应获取操作时，while循环将终止。

tryReleaseShared将增加许可计数，这可能会解除等待中线程阻塞状态，并且不断尝试直到更新操作成功。

tryReleaseShared的返回值表示在这次释放操作中解除了其他线程的阻塞。

CountDownLatch使用AQS的方式与Semaphore很相似：在同步状态中保存的是当前的计数值。countDown方法调用release，从而导致计数值递减，并且当计数值为0时，解除所有等待线程的阻塞，await调用acquire方法，当计数器为0时，acquire立即返回，否则将阻塞。

14.6.3 FutureTask

Future.get的语义非常类似与闭锁的语义——如果发生了某个事件（由于FutureTask表示的任务执行完成或被取消），那么线程就可以恢复执行，否则这些线程将停留在队列中直到事件发生。

FutureTask中AQS的同步状态被用来保存任务的状态。例如，正在运行，已完成或已取消。FutureTask还维护一些额外的状态变量，用来保存计算或抛出的异常。

此外，它还维护了一个引用，指向正在执行任务的线程（如果它当前处于运行状态），因而如果任务取消，线程就会中断

14.6.4 ReentrantReadWriteLock

ReadWriteLock接口表示存在两个锁：一个读取锁和一个写入锁，但在基于AQS实现的ReentrantReadWriteLock中，单个AQS子类将同时管理读取加锁和写入加锁。

ReentrantReadWriteLock分别使用了两个16位的状态来表示写入锁的计数和读取锁的计数。

在读取锁啥好过你的操作将使用共享的获取方法和释放方法，在写入锁上的操作将使用独占的获取方法与释放方法。

AQS在内部维护一个等待线程队列，其中记录了某个线程请求的是独占访问还是共享访问。

在ReentrantReadWriteLock中，当锁可用时，如果位于队列头部的线程执行写入操作，那么线程会获得这个锁，

如果位于队列头部的线程执行读取访问，那么队列中在第一个写入线程之前的所有线程都将获得这个锁。

小结

要实现一个依赖状态的类——如果没有满足依赖状态的前提条件，那么这个类的方法必须阻塞，那么最好的方式是基于现有的库类来构建。

有时候现有的类库不能提供足够的功能，可以使用内置的条件队列，显式的Condition对象或AQS来构建自己的同步器。