看一遍就能明白的Java并发基石AQS

并发之父

Doug Lea(小名：李二狗)

俗话说的好，生平不识Doug Lea，学懂并发也枉然，哈哈哈！

        在介绍AQS之前我们先回忆一下我们脑海中学过的一些并发工具包，如CountDownLatch、ReentrantLock、Semphore和ReentrantReadWriteLock等，从它们的源码中，我们可以发现它们都定义了一个内部类，用来实现AbstractQueuedSynchronizer类，AbstractQueuedSynchronizer的简称就是我们今天要介绍的主角AQS。

一、原理解析

        在AQS中，维护这一个表示共享资源加锁情况的变量voliatile int state，以及一个FIFO的线程阻塞队列(称为CLH队列)，当多个线程并发访问到共享资源时，如果共享资源已经被某个线程加了锁，那么其他线程将进入CLH队列中等待。

state表示共享资源被加锁次数。state = 0 表示没有被加锁；state >=1 表示被加锁。访问方式如下：

并发线程在访问共享资源时都会使用以下一种或两种方式加锁：

Exclusive：独占式，同一时间内只能有一个线程访问资源，如ReentrantLock采用的是独占方式。

Share：共享方式，同一时间内允许有多个线程访问资源，如CountDownLatch、Semphore等。

二、源码解析

废话不多说，直接奉上AbstractQueuedSynchronizer部分源码：

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {


   static final class Node {
       // 共享模式的Node
       static final Node SHARED = new Node();
       // 独占模式的Node
       static final Node EXCLUSIVE = null;
       //Node的等待状态，共有CANCELLED、SIGNAL、CONDITION、PROPAGATE四种状态。waitStatus的初始状态为 0
       volatile int waitStatus;


       /**
        * 失效状态：如果在CLH中的线程等待超时或被中断，就需要从CLH中取消该Node节点，并将该Node的waitStatus设置为CANCELLED
        * 注意：CANCELLED的值为1，也是所有状态中，唯一一个大于0的值
        */
       static final int CANCELLED = 1;
       /**
        * 如果某一个Node的前驱节点正在加锁并占用资源，当这个前驱节点释放锁后就会唤醒waitStatus = SIGNAL的Node节点。也就是说
        * waitStatus = SIGNAL 的 Node就是下一个能够占用资源的Node节点
        */
       static final int SIGNAL = -1;
       /**
        * waitStatus = condition 的Node中的线程正在等待某一个Condition，当其他线程调用了该Condition的signal()方法后，就会将该
        * Node节点从等待队列转移到同步队列，等待获取同步锁
        */
       static final int CONDITION = -2;
       /**
        * 在共享模式中，waitStatus = propagate 的Node中的线程处于可运行状态
        */
       static final int PROPAGATE = -3;


  }
   // 头结点
   private transient volatile Node head;


   // 尾结点
   private transient volatile Node tail;


   // 资源状态，即被加锁的次数
   private volatile int state;


   protected final int getState() {
       return state;
  }


   protected final void setState(int newState) {
       state = newState;
  }


   protected final boolean compareAndSetState(int expect, int update) {
       return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
  }
}

CLH中的Node节点是如何等待，移动到队首的呢？

       在AQS源码中还提供了acquireQueued()方法和acquire()方法。当线程A访问资源失败，就会以Node的形式(记为Node-A)被加入到CLH队尾，acquireQueued()方法会判断Node-A是不是CLH中的第二个节点(第二个执行状态)，如果是就通过"自旋"不断的去尝试占用资源；如果不是，则安心的处于等待状态，直到自己前移到第二个位置。

acquireQueued()方法的相关源码如下：

final boolean acquireQueued(final Node node, int arg) {
   //如果获取成功返回false,如果失败返回true
   boolean failed = true;
   try {
       //等待过程中，是否被中断过
       boolean interrupted = false;
       //自旋
       for (; ; ) {
           //获取当前节点的前驱节点
           final Node p = node.predecessor();
           //如果前驱节点是头节点，则自己就是第二个节点，通过自旋不断的尝试获取资源(tryAcquire(arg))
           if (p == head && tryAcquire(arg)) {
               //获取成功，将自己设为头结点
               setHead(node);
               p.next = null; // help GC
               failed = false;
               return interrupted;
          }
           //如果当前节点不在前两个位置，则放心等待，直到被唤醒(unpark())
           if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()){
               //当前线程在等待中被中断，则设为true
               interrupted = true;
          }
      }
  } finally {
       if (failed)
           cancelAcquire(node);
  }
}


/*
这个方法总体来说：如果node不在前面的两个位置，那么就可以安心的等待下去，但是有一些特殊情况，比如：
前面的某些node是无效的，那么当轮到这些无效的node占用资源时，这些node将会放弃占用，因此CLH会迅速切换到下一个node。
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
   //获取前驱节点的等待状态
   int ws = pred.waitStatus;
   //如果前驱节点的状态是SIGNAL，则表示这个前驱节点在前移到第一位时(即占用资源状态)时，告知一下自己。这样当前节点就可以放心的休息了。
   if (ws == Node.SIGNAL)
       return true;
   //如果前驱节点的状态是大于0，即CANCELLED时，则当前节点就一直往前移，直到移动到一个真正等待的节点后面
   if (ws > 0) {
       do {
           node.prev = pred = pred.prev;
      } while (pred.waitStatus > 0);
       pred.next = node;
  } else {
    //如果前驱节点是正常等待状态，就把前驱节点的状态设置为SIGNAL
       compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
   return false;
}


//当前节点放心等待的方法
private final boolean parkAndCheckInterrupt() {
   //调用park()方法，让当前线程进入waiting状态
   LockSupport.park(this);
   //如果被唤醒，还要检查是否是被中断的。(1.正常调用unpark()被唤醒；2.被interrupt()方法中断)
   return Thread.interrupted();
}

      再来看看AQS中独占模式下线程获取共享资源的顶层方法acquire(),源码如下：

public final void acquire(int arg) {
  //尝试独占资源的流程如下：
  // 1.如果执行tryAcquire()方法后返回的结果为true，则独占成功，直接结束。
  // 2.addWaiter()：如果独占失败，就将该线程的node标记为独占模式，并加入到CLH的队尾
  // 3.因为是独占模式，当前node独占失败后会在CLH中等待，知道等待结束成功独占资源。
  // 4.如果node独占失败，并且在CLH中等待的过程中出现了中断，就还需要执行selfInterrupt()
   if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
       selfInterrupt();
}

        与acquire()方法相反的release()就是释放锁的顶层方法，每执行一次release(1)，就会将共享资源上加锁的次数减1.如果state=0，就说明共享资源彻底释放了，此时就会唤醒CLH中下一个等待的线程。其源码如下：

public final boolean release(int arg) {
   //释放加在共享资源上的arg把锁，如果释放成功，则返回true
   if (tryRelease(arg)) {
       //获取CLH中的头结点
       Node h = head;
       //唤醒下一个结点
       if (h != null && h.waitStatus != 0)
           unparkSuccessor(h);
       return true;
  }
   return false;
}


//唤醒CLH中下一个“正常等待状态的”结点
private void unparkSuccessor(Node node) {
//获取正在占用资源的当前节点状态
   int ws = node.waitStatus;
   //将结点的状态通过CAS恢复成初始值0
   if (ws < 0)
       compareAndSetWaitStatus(node, ws, 0);
//获取下一个结点，也就是即将被唤醒的节点
   Node s = node.next;
   //如果下一个结点为null或是失效状态，就将下一个结点设置为null；并从头到尾往前遍历，直到找到一个处于正常等待状态的节点，进行唤醒。
   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);
}

以上是对独占模式加锁解锁源码的解读。对于共享模式加锁和解锁的方法分别是acquireShared(int)和releaseShared(int)，思路大致相同，读者可以自行尝试阅读。

下面附上部分ReentrantReadWriteLock源码，因为ReentrantReadWriteLock独占模式和共享模式这两种方式都实现了，如下：

• tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。

• tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。

• isHeldExclusively()：该线程是否正在独占资源。只有用到condition才需要去实现它。

• tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。

• tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。

  
  

• /**
  
  
  * @since 1.5
  * @author Doug Lea
  */
  public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable {
  
  
     private final ReentrantReadWriteLock.ReadLock readerLock;
     private final ReentrantReadWriteLock.WriteLock writerLock;
     final Sync sync;
  
  
  // 默认是非公平锁
     public ReentrantReadWriteLock() {
         this(false);
    }
  
  
     // new对象时就生成了两把锁，一把读锁，一把写锁
     public ReentrantReadWriteLock(boolean fair) {
         sync = fair ? new FairSync() : new NonfairSync();
         readerLock = new ReadLock(this);
         writerLock = new WriteLock(this);
    }
  
  
     public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
     public ReentrantReadWriteLock.ReadLock  readLock() { return readerLock; }
  
  
     abstract static class Sync extends AbstractQueuedSynchronizer {
  
  
         // 独占方式加锁
         protected final boolean tryAcquire(int acquires) {
             Thread current = Thread.currentThread();
             int c = getState();
             int w = exclusiveCount(c);
             if (c != 0) {
                 // (Note: if c != 0 and w == 0 then shared count != 0)
                 if (w == 0 || current != getExclusiveOwnerThread())
                     return false;
                 if (w + exclusiveCount(acquires) > MAX_COUNT)
                     throw new Error("Maximum lock count exceeded");
                 // Reentrant acquire
                 setState(c + acquires);
                 return true;
            }
             if (writerShouldBlock() ||
                 !compareAndSetState(c, c + acquires))
                 return false;
             setExclusiveOwnerThread(current);
             return true;
        }
  
  
         // 独占方式解锁
         protected final boolean tryRelease(int releases) {
             if (!isHeldExclusively())
                 throw new IllegalMonitorStateException();
             int nextc = getState() - releases;
             boolean free = exclusiveCount(nextc) == 0;
             if (free)
                 setExclusiveOwnerThread(null);
             setState(nextc);
             return free;
        }
         
     // 共享方式加锁
         protected final int tryAcquireShared(int unused) {
             Thread current = Thread.currentThread();
             int c = getState();
             if (exclusiveCount(c) != 0 &&
                 getExclusiveOwnerThread() != current)
                 return -1;
             int r = sharedCount(c);
             if (!readerShouldBlock() &&
                 r < MAX_COUNT &&
                 compareAndSetState(c, c + SHARED_UNIT)) {
                 if (r == 0) {
                     firstReader = current;
                     firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                     firstReaderHoldCount++;
                } else {
                     HoldCounter rh = cachedHoldCounter;
                     if (rh == null || rh.tid!= getThreadId(current))
                         cachedHoldCounter = rh = readHolds.get();
                     else if (rh.count == 0)
                         readHolds.set(rh);
                     rh.count++;
                }
                 return 1;
            }
             return fullTryAcquireShared(current);
        }
  
  
         // 共享方式解锁
         protected final boolean tryReleaseShared(int unused) {
             Thread current = Thread.currentThread();
             if (firstReader == current) {
                 if (firstReaderHoldCount == 1)
                     firstReader = null;
                 else
                     firstReaderHoldCount--;
            } else {
                 HoldCounter rh = cachedHoldCounter;
                 if (rh == null || rh.tid != getThreadId(current))
                     rh = readHolds.get();
                 int count = rh.count;
                 if (count <= 1) {
                     readHolds.remove();
                     if (count <= 0)
                         throw unmatchedUnlockException();
                }
                 --rh.count;
            }
             for (;;) {
                 int c = getState();
                 int nextc = c - SHARED_UNIT;
                 if (compareAndSetState(c, nextc))
                     return nextc == 0;
            }
        }
         
     // 判断当前线程是否独占共享资源
         protected final boolean isHeldExclusively() {
             return getExclusiveOwnerThread() == Thread.currentThread();
        }
  
  
  }
  
  
  
  
  
  
  
  
  

以上就是本次内容，如您有不准确之处欢迎留言交流。