深入浅出 Java Concurrency (9): 锁机制 part 4

  来源:深入浅出 Java Concurrency                                                 作者:饭饭泛


本小节介绍锁释放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操作是这样的:

    1. 判断持有锁的线程是否是当前线程,如果不是就抛出IllegalMonitorStateExeception(),因为一个线程是不能释放另一个线程持有的锁(否则锁就失去了意义)。否则进行2。

    2. 将AQS状态位减少要释放的次数(对于独占锁而言总是1),如果剩余的状态位0(也就是没有线程持有锁),那么当前线程就是最后一个持有锁的线程,清空AQS持有锁的独占线程。进行3。

    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队列中剔除  
      //这里并没有从head->tail开始寻找,而是从tail->head寻找最后一个有效节点。
      //解释在这里 http://www.blogjava.net/xylz/archive/2010/07/08/325540.html#377512

       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);
}

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

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

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.waitsignal对应于Object.notifysignalAll对应于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 {

   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()操作是安装如下步骤进行的:

    1. 将当前线程加入Condition锁队列。特别说明的是,这里不同于AQS的队列,这里进入的是Condition的FIFO队列。后面会具体谈到此结构。进行2。

    2. 释放锁。这里可以看到将锁释放了,否则别的线程就无法拿到锁而发生死锁。进行3。

    3. 自旋(while)挂起,直到被唤醒或者超时或者CACELLED等。进行4。

    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队列。

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



本文链接:http://www.hihubs.com/article/199

关键字:深入浅出 Java Concurrency (9): 锁机制 part 4

若无特别注明,文章皆为Hubs'm原创,转载请注明出处...O(∩_∩)O


微信沟通