《深入浅出 Java Concurrency》—锁机制(四) 锁释放与条件变量 (Lock.unlock And Condition)

转自：http://www.blogjava.net/xylz/archive/2010/07/08/325540.html

本小节介绍锁释放Lock.unlock()。

Release/TryRelease

unlock操作实际上就调用了AQS 的release操作，释放持有的锁。

public final boolean release(int arg) {

    if (tryRelease(arg)) {

        Node h = head;

        if (h != null && h.waitStatus != 0)

            unparkSuccessor(h);

        return true;

    }

    return false;

}

前面提到过tryRelease(arg) 操作，此操作里面总是尝试去释放锁，如果成功，说明锁确实被当前线程持有，那么就看AQS 队列中的头结点是否为空并且能否被唤醒，如果可以的话就唤醒继任节点（下一个非CANCELLED节点，下面会具体分析）。

对于独占锁而言，java.util.concurrent.locks.ReentrantLock.Sync.tryRelease(int)展示了如何尝试释放锁(tryRelease )操作。

protected final boolean tryRelease(int releases) {

    int c = getState() - releases;

    if (Thread.currentThread() != getExclusiveOwnerThread())

        throw new IllegalMonitorStateException();

    boolean free = false;

    if (c == 0) {

        free = true;

        setExclusiveOwnerThread(null);

    }

    setState(c);

    return free;

}

整个tryRelease 操作是这样的：
 
判断持有锁的线程是否是当前线程，如果不是就抛出IllegalMonitorStateExeception()，因为一个线程是不能释放另一个线程持有的锁（否则锁就失去了意义）。否则进行2。
将AQS状态位减少要释放的次数（对于独占锁而言总是1），如果剩余的状态位0（也就是没有线程持有锁），那么当前线程就是最后一个持有锁的线程，清空AQS持有锁的独占线程。进行3。
将剩余的状态位写回AQS，如果没有线程持有锁就返回true，否则就是false。

参考上一节的分析就可以知道，这里c==0决定了是否完全释放了锁。由于ReentrantLock 是可重入锁，因此同一个线程可能多重持有锁，那么当且仅当最后一个持有锁的线程释放锁是才能将AQS中持有锁的独占线程清空，这样接下来的操作才需要唤醒下一个需要锁的AQS 节点（Node），否则就只是减少锁持有的计数器，并不能改变其他操作。

当tryRelease 操作成功后（也就是完全释放了锁），release操作才能检查是否需要唤醒下一个继任节点。这里的前提是AQS 队列的头结点需要锁(waitStatus!=0 )，如果头结点需要锁，就开始检测下一个继任节点是否需要锁操作。

在上一节中说道acquireQueued 操作完成后（拿到了锁），会将当前持有锁的节点设为头结点，所以一旦头结点释放锁，那么就需要寻找头结点的下一个需要锁的继任节点，并唤醒它。

private void unparkSuccessor(Node node) {

        //此时node是需要是需要释放锁的头结点

        //清空头结点的waitStatus，也就是不再需要锁了

        compareAndSetWaitStatus(node, Node.SIGNAL, 0);

        //从头结点的下一个节点开始寻找继任节点，当且仅当继任节点的waitStatus<=0才是有效继任节点，否则将这些waitStatus>0（也就是CANCELLED的节点）从AQS队列中剔除  

        Node s = node.next;

        if (s == null || s.waitStatus > 0) {

            s = null;

            for (Node t = tail; t != null && t != node; t = t.prev)

                if (t.waitStatus <= 0)

                    s = t;

        }

        //如果找到一个有效的继任节点，就唤醒此节点线程

        if (s != null)

            LockSupport.unpark(s.thread);

    }

这里再一次把acquireQueued 的过程找出来。对比unparkSuccessor ，一旦头节点的继任节点被唤醒，那么继任节点就会尝试去获取锁（在acquireQueued 中node就是有效的继任节点，p就是唤醒它的头结点），如果成功就会将头结点设置为自身，并且将头结点的前任节点清空，这样前任节点（已经过时了）就可以被GC释放了。

final boolean acquireQueued(final Node node, int arg) {

    try {

        boolean interrupted = false;

        for (;;) {

            final Node p = node.predecessor();

            if (p == head && tryAcquire(arg)) {

                setHead(node);

                p.next = null; // help GC

                return interrupted;

            }

            if (shouldParkAfterFailedAcquire(p, node) &&

                parkAndCheckInterrupt())

                interrupted = true;

        }

    } catch (RuntimeException ex) {

        cancelAcquire(node);

        throw ex;

    }

}

在setHead 中，将头结点的前任节点清空并且将头结点的线程清空就是为了更好的GC，防止内存泄露。

private void setHead(Node node) {

    head = node;

    node.thread = null;

    node.prev = null;

}

对比lock()操作，unlock()操作还是比较简单的，主要就是释放响应的资源，并且唤醒AQS 队列中有效的继任节点。这样所就按照请求的顺序去尝试获取锁了。

整个lock()/unlock()过程完成了，我们再回头看公平锁(FairSync)和非公平锁(NonfairSync)。

公平锁和非公平锁只是在获取锁的时候有差别，其它都是一样的。

final void lock() {

    if (compareAndSetState(0, 1))

        setExclusiveOwnerThread(Thread.currentThread());

    else

        acquire(1);

}

在上面非公平锁的代码中总是优先尝试当前是否有线程持有锁，一旦没有任何线程持有锁，那么非公平锁就霸道的尝试将锁“占为己有”。如果在抢占锁的时候失败就和公平锁一样老老实实的去排队。

也即是说公平锁和非公平锁只是在入AQS 的CLH 队列之前有所差别，一旦进入了队列，所有线程都是按照队列中先来后到的顺序请求锁。

Condition

条件变量很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。

条件（也称为条件队列   或条件变量 ）为线程提供了一个含义，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”）。因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是：以原子方式   释放相关的锁，并挂起当前线程，就像 

Object.wait

   做的那样。

上述API说明表明条件变量需要与锁绑定，而且多个Condition需要绑定到同一锁上。前面的Lock 中提到，获取一个条件变量的方法是Lock.newCondition() 。

void await() throws InterruptedException;

void awaitUninterruptibly();

long awaitNanos(long nanosTimeout) throws InterruptedException;

boolean await(long time, TimeUnit unit) throws InterruptedException;

boolean awaitUntil(Date deadline) throws InterruptedException;

void signal();

void signalAll();

以上是Condition 接口定义的方法，await* 对应于Object.wait ，signal 对应于Object.notify ，signalAll 对应于Object.notifyAll 。特别说明的是Condition 的接口改变名称就是为了避免与Object中的wait/notify/notifyAll 的语义和使用上混淆，因为Condition同样有wait/notify/notifyAll 方法。

每一个Lock 可以有任意数据的Condition 对象，Condition 是与Lock 绑定的，所以就有Lock 的公平性特性：如果是公平锁，线程为按照FIFO的顺序从Condition.await 中释放，如果是非公平锁，那么后续的锁竞争就不保证FIFO顺序了。

一个使用Condition实现生产者消费者的模型例子如下。

package xylz.study.concurrency.lock;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

public class ProductQueue<T> {

    private final T[] items;

    private final Lock lock = new ReentrantLock();

    private Condition notFull = lock.newCondition();

    private Condition notEmpty = lock.newCondition();

    //

    private int head, tail, count;

    public ProductQueue(int maxSize) {

        items = (T[]) new Object[maxSize];

    }

    public ProductQueue() {

        this(10);

    }

    public void put(T t) throws InterruptedException {

        lock.lock();

        try {

            while (count == getCapacity()) {

                notFull.await();

            }

            items[tail] = t;

            if (++tail == getCapacity()) {

                tail = 0;

            }

            ++count;

            notEmpty.signalAll();

        } finally {

            lock.unlock();

        }

    }

    public T take() throws InterruptedException {

        lock.lock();

        try {

            while (count == 0) {

                notEmpty.await();

            }

            T ret = items[head];

            items[head] = null;//GC

            //

            if (++head == getCapacity()) {

                head = 0;

            }

            --count;

            notFull.signalAll();

            return ret;

        } finally {

            lock.unlock();

        }

    }

    public int getCapacity() {

        return items.length;

    }

    public int size() {

        lock.lock();

        try {

            return count;

        } finally {

            lock.unlock();

        }

    }

}

在这个例子中消费take() 需要 队列不为空，如果为空就挂起（await() ），直到收到notEmpty 的信号；生产put() 需要队列不满，如果满了就挂起（await() ），直到收到notFull 的信号。

可能有人会问题，如果一个线程lock() 对象后被挂起还没有unlock ，那么另外一个线程就拿不到锁了（lock() 操作会挂起），那么就无法通知(notify )前一个线程，这样岂不是“死锁”了？

 

await* 操作

上一节中说过多次ReentrantLock 是独占锁，一个线程拿到锁后如果不释放，那么另外一个线程肯定是拿不到锁，所以在lock.lock() 和lock.unlock() 之间可能有一次释放锁的操作（同样也必然还有一次获取锁的操作）。我们再回头看代码，不管take() 还是put() ，在进入lock.lock() 后唯一可能释放锁的操作就是await() 了。也就是说await() 操作实际上就是释放锁，然后挂起线程，一旦条件满足就被唤醒，再次获取锁！

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)

        unlinkCancelledWaiters();

    if (interruptMode != 0)

        reportInterruptAfterWait(interruptMode);

}

上面是await() 的代码片段。上一节中说过，AQS 在获取锁的时候需要有一个CHL 的FIFO队列，所以对于一个Condition.await() 而言，如果释放了锁，要想再一次获取锁那么就需要进入队列，等待被通知获取锁。完整的await()操作是安装如下步骤进行的：
 
将当前线程加入Condition 锁队列。特别说明的是，这里不同于AQS 的队列，这里进入的是Condition的FIFO队列。后面会具体谈到此结构。进行2。
释放锁。这里可以看到将锁释放了，否则别的线程就无法拿到锁而发生死锁。进行3。
自旋(while)挂起，直到被唤醒或者超时或者CACELLED等。进行4。
获取锁(acquireQueued )。并将自己从Condition 的FIFO队列中释放，表明自己不再需要锁（我已经拿到锁了）。

这里再回头介绍Condition 的数据结构。我们知道一个Condition 可以在多个地方被await*() ，那么就需要一个FIFO的结构将这些Condition 串联起来，然后根据需要唤醒一个或者多个（通常是所有）。所以在Condition 内部就需要一个FIFO的队列。

private transient Node firstWaiter;

private transient Node lastWaiter;

上面的两个节点就是描述一个FIFO的队列。我们再结合前面 提到的节点（Node）数据结构 。我们就发现Node.nextWaiter 就派上用场了！nextWaiter 就是将一系列的Condition.await* 串联起来组成一个FIFO的队列。

 

signal/signalAll 操作

await*() 清楚了，现在再来看signal/signalAll 就容易多了。按照signal/signalAll 的需求，就是要将Condition.await*()中FIFO队列中第一个Node 唤醒（或者全部Node ）唤醒。尽管所有Node 可能都被唤醒，但是要知道的是仍然只有一个线程能够拿到锁，其它没有拿到锁的线程仍然需要自旋等待，就上上面提到的第4步(acquireQueued)。

private void doSignal(Node first) {

    do {

        if ( (firstWaiter = first.nextWaiter) == null)

            lastWaiter = null;

        first.nextWaiter = null;

    } while (!transferForSignal(first) &&

             (first = firstWaiter) != null);

}

private void doSignalAll(Node first) {

    lastWaiter = firstWaiter  = null;

    do {

        Node next = first.nextWaiter;

        first.nextWaiter = null;

        transferForSignal(first);

        first = next;

    } while (first != null);

}

上面的代码很容易看出来，signal 就是唤醒Condition 队列中的第一个非CANCELLED节点线程，而signalAll就是唤醒所有非CANCELLED节点线程。当然了遇到CANCELLED线程就需要将其从FIFO队列中剔除。

final boolean transferForSignal(Node node) {

    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))

        return false;

    Node p = enq(node);

    int c = p.waitStatus;

    if (c > 0 || !compareAndSetWaitStatus(p, c, Node.SIGNAL))

        LockSupport.unpark(node.thread);

    return true;

}

上面就是唤醒一个await*() 线程的过程，根据前面的小节介绍的，如果要unpark 线程，并使线程拿到锁，那么就需要线程节点进入AQS 的队列。所以可以看到在LockSupport.unpark 之前调用了enq(node) 操作，将当前节点加入到AQS 队列。

整个锁机制的原理就介绍完了，从下一节开始就进入了锁机制的应用了。