队列同步器AQS-AbstractQueuedSynchronizer 原理分析

private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;
Node
ConditionObject
protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
      return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

//独占模式
protected boolean tryAcquire(int arg);
protected boolean tryRelease(int arg);
//共享模式
protected int tryAcquireShared(int arg);
protected boolean tryReleaseShared(int arg);
//判断当前线程是否排它持有state
protected boolean isHeldExclusively()；
//以上方法在AQS中默认实现为throw new UnsupportedOperationException();

protected final int getState()
protected final void setState(int newState)
protected final boolean compareAndSetState(int expect, int update)

* tryAcquire
* tryRelease
* tryAcquireShared
* tryReleaseShared
* isHeldExclusively

public abstract class AbstractOwnableSynchronizer
    implements java.io.Serializable {
    /** Use serial ID even though all fields transient. */
    private static final long serialVersionUID = 3737899427754241961L;
    protected AbstractOwnableSynchronizer() { }
     //独占模式同步器的当前所有者-线程
    private transient Thread exclusiveOwnerThread;
     //设置当前独有访问的线程。如果参数为null，表示没有线程接入。
     //参数并不需要额外强加任何同步或者volatile类型修饰
    protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
    }
     //返回被setExclusiveOwnerThread方法最近set的线程，如果没有set，返回null。
     //This method does not otherwise impose any synchronization or {@code volatile} field accesses.
    protected final Thread getExclusiveOwnerThread() {
        return exclusiveOwnerThread;
    }
}

Acquire:
    while (!tryAcquire(arg)) {
       enqueue thread if it is not already queued;
       possibly block current thread;
     }
 Release:
     if (tryRelease(arg))
       unblock the first queued thread

public class Mutex implements Lock, java.io.Serializable {
   // Our internal helper class
   private static class Sync extends AbstractQueuedSynchronizer {
     // Reports whether in locked state
     protected boolean isHeldExclusively() {
       return getState() == 1;
     }
     // Acquires the lock if state is zero
     public boolean tryAcquire(int acquires) {
       assert acquires == 1; // Otherwise unused
       if (compareAndSetState(0, 1)) {
       //AQS父类AOS的方法
         setExclusiveOwnerThread(Thread.currentThread());
         return true;
       }
       return false;
     }
     // Releases the lock by setting state to zero
     protected boolean tryRelease(int releases) {
       assert releases == 1; // Otherwise unused
       if (getState() == 0) throw new IllegalMonitorStateException();
       setExclusiveOwnerThread(null);
       setState(0);
       return true;
     }
     // Provides a Condition
     Condition newCondition() { return new ConditionObject(); }
     // Deserializes properly
     private void readObject(ObjectInputStream s)
         throws IOException, ClassNotFoundException {
       s.defaultReadObject();
       setState(0); // reset to unlocked state
     }
   }
   // The sync object does all the hard work. We just forward to it.
   private final Sync sync = new Sync();
//lock  unlock的本质是使用继承自AQS的子类的acquire和release方法
   public void lock()                { sync.acquire(1); }
   public boolean tryLock()          { return sync.tryAcquire(1); }
   public void unlock()              { sync.release(1); }
   public Condition newCondition()   { return sync.newCondition(); }
   public boolean isLocked()         { return sync.isHeldExclusively(); }
   public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
   public void lockInterruptibly() throws InterruptedException {
     sync.acquireInterruptibly(1);
   }
   public boolean tryLock(long timeout, TimeUnit unit)
       throws InterruptedException {
     return sync.tryAcquireNanos(1, unit.toNanos(timeout));
   }
 }

import java.util.concurrent.CyclicBarrier;
public class TestMutex {
//这里使用栅栏，可以reset count.
    private static CyclicBarrier barrier = new CyclicBarrier(31);
    private static int a = 0;
    private static  Mutex mutex = new Mutex();
    public static void main(String []args) throws Exception {
        //说明:我们启用30个线程，每个线程对i自加10000次，同步正常的话，最终结果应为300000；
        //未加锁前
        for(int i=0;i<30;i++){
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    for(int i=0;i<10000;i++){
                        increment1();//没有同步措施的a++；
                    }
                    try {
                        barrier.await();//等30个线程累加完毕
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
            t.start();
        }
        barrier.await();
        System.out.println("加锁前，a="+a);
        //加锁后
        barrier.reset();//重置CyclicBarrier
        a=0;
        for(int i=0;i<30;i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for(int i=0;i<10000;i++){
                        increment2();//a++采用Mutex进行同步处理
                    }
                    try {
                        barrier.await();//等30个线程累加完毕
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        barrier.await();
        System.out.println("加锁后，a="+a);
    }
    /**
     * 没有同步措施的a++
     * @return
     */
    public static void increment1(){
        a++;
    }
    /**
     * 使用自定义的Mutex进行同步处理的a++
     */
    public static void increment2(){
        mutex.lock();
        a++;
        mutex.unlock();
    }
}

加锁前，a=204889
加锁后，a=300000

public class BooleanLatch {
  private static class Sync extends AbstractQueuedSynchronizer {
        boolean isSignalled() { return getState() != 0; }
        protected int tryAcquireShared(int ignore) {
          return isSignalled() ? 1 : -1;
        }
        protected boolean tryReleaseShared(int ignore) {
          setState(1);
          return true;
        }
      }
      private final Sync sync = new Sync();
      public boolean isSignalled() { return sync.isSignalled(); }
      public void signal()         { sync.releaseShared(1); }
      public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
          }
}

//返回同步状态的当前值
protected final int getState() {  
        return state;
}
 // 设置同步状态的值
protected final void setState(int newState) { 
        state = newState;
}
//原子地（CAS操作）将同步状态值设置为给定值update如果当前同步状态的值等于expect（期望值）
protected final boolean compareAndSetState(int expect, int update) {
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

isHeldExclusively()//该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int)//独占方式。尝试获取资源，成功则返回true，失败则返回false。
tryRelease(int)//独占方式。尝试释放资源，成功则返回true，失败则返回false。
tryAcquireShared(int)
//共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；
//正数表示成功，且有剩余资源。
tryReleaseShared(int)
//共享方式。尝试释放资源，成功则返回true，失败则返回false。

     +------+  prev +-----+       +-----+
     head |      | <---- |     | <---- |     |  tail
     +------+       +-----+       +-----+

static final class Node {
        //标记一个正在共享模式中等待的节点
        static final Node SHARED = new Node();
        //标记一个正在独占模式中等待的节点
        static final Node EXCLUSIVE = null;
  //waitStatus 值去声明线程被cancelle
        static final int CANCELLED =  1;
  //waitStatus 值 去声明继任者线程需要unparking 
        static final int SIGNAL    = -1;
     //waitStatus 值 去声明线程等待一个condition 
        static final int CONDITION = -2;
         //waitStatus  值声明下一个acquireShared 应该无条件传播
        static final int PROPAGATE = -3;
/**
* waitStatus取值的几种情况：
* 
*SIGNAL:     
当前结点的继任者是阻塞的(通过park方法)，所以在当前结点被释放或者取消时必须unpark它的继任者。
为了避免竞争，获取方法必须首先指示它们需要一个signal，然后重试原子获取，然后在失败时阻塞。
* 
*CANCELLED:  
这个结点由于超时或者中断被取消。节点永远离不开这个状态。
特别是，具有取消节点的线程不会再次阻塞。
* 
*CONDITION:  
此节点当前处于条件队列中。
在传输之前，它不会被用作同步队列节点，
此时状态将被设置为0(这里使用这个值与字段的其他用途无关，但是简化了技术)。
* 
* PROPAGATE:  releaseShared 应该被传播给其他结点。 
* 这在doReleaseShared中设置（仅针对头节点）以确保传播继续，即使其他操作已经介入。
* 
 0:   其他情况,一般为结点初始化值
*
值用数字来表示以简化使用。非负值意味着节点不需要signal。
所以，大多数代码不需要检查特定的值，只是为了签名
对于正常同步结点该字段被初始化为0，对于condition 结点则为CONDITION。
使用CAS算法修改(or when possible, unconditional volatile writes)
        volatile int waitStatus;
         //链接到当前节点/线程依赖于检查waitStatus的前驱节点.
         //入队时分配，出队时置null(为了GC)。
         //此外，在取消前置节点时，我们在找到未取消的节点时短路，这总是存在的，因为head节点从未被取消：节点仅由于成功acquire而变为前置节点。
         //被取消的线程永远不会成功acquiring，并且线程只取消自身，而不是任何其他节点。
        volatile Node prev;
         //链接到当前节点/线程在释放时唤醒的后继节点。
         //在排队期间分配，在绕过取消的前辈结点时进行调整，在退出队列时取消（为了GC）。
         //入队操作直到附加之后才分配前一个结点的"next"字段，所以看到null的"next"字段并不一定意味着节点在队列末尾。
         //但是，如果"next"字段看起来是空的，我们可以从尾部扫描"prev's"来进行双重检查。
         //取消节点的"next"字段被设置为指向节点本身，而不是null，以使isOnSyncQueue更容易操作。
        volatile Node next;
        //排队节点的线程。构造时初始化，out后失效。
        volatile Thread thread;
    /*     链接到下个等待状态的结点，或着特殊值SHARED。
         因为条件队列只有在独占模式中保持时才会被访问，所以我们只需要一个简单的链接队列来在节点等待Condition时保持节点。
         然后，他们被转移到队列重新获取。
         因为Condition只能是排他的，所以我们使用特殊值来指示共享模式。*/
        Node nextWaiter;
        //如果结点在共享模式中等待，返回true
        final boolean isShared() {
            return nextWaiter == SHARED;
        }
        //返回"前辈"结点，如果为null则抛出NullPointerException 。
      //  可以省去NULL检查，但存在以帮助VM
        final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }
        Node() {    // Used to establish initial head or SHARED marker
        }
        Node(Thread thread, Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }
        Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

static final int CANCELLED =  1;
static final int SIGNAL    = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;

public final void acquire(int arg);
final boolean acquireQueued(final Node node, int arg)
public final void acquireInterruptibly(int arg);
public final void acquireShared(int arg);
public final void acquireSharedInterruptibly(int arg);
protected boolean tryAcquire(int arg); 
protected int tryAcquireShared(int arg);
public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException;
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException;

//独占模式中获取锁的方法，排队状态时可能多次阻塞和非阻塞。通常用来实现lock方法
public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

 private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        //判断pred是否为null
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //这里走入队操作
        enq(node);
        return node;

private Node enq(final Node node) {
  //死循环
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
            // 还记得head 和tail 结点都是懒初始化的吧
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
              // node的prev设置为t
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                  //t.next 设置为node  双向链表常用操作
                    t.next = node;
                    return t;
                }
            }
        }
    }

//Acquires in exclusive uninterruptible mode for thread already in
// queue. Used by condition wait methods as well as acquire.
final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            //死循环
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

 /**
     * Sets head of queue to be node, thus dequeuing. Called only by
     * acquire methods.  Also nulls out unused fields for sake of GC
     * and to suppress unnecessary signals and traversals.
     *
     * @param node the node
     */
    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

/**检查并更新一个acquire失败的结点的状态。返回线程是否应该被阻塞。
在所有的acquire的循环中是主要的singnal控制。
Requires that pred == node.prev.
 * @param pred node's predecessor holding status
* @param node the node 
*/
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)// -1
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {// 1 cancelled
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

public final boolean release(int arg); 
public final boolean releaseShared(int arg)
protected boolean tryRelease(int arg); 
protected boolean tryReleaseShared(int arg);

//Releases in exclusive mode.  Implemented by unblocking one or
//more threads if {@link #tryRelease} returns true.
//This method can be used to implement method {@link Lock#unlock}.
public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

//Wakes up node's successor, if one exists
private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        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);
    }

 final boolean isOnSyncQueue(Node node) {
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        if (node.next != null) // If has successor, it must be on queue
            return true;
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
        return findNodeFromTail(node);
    }

 protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();
 }
返回值情况如下：
1.当返回值大于0时，表示获取同步状态成功，同时还有剩余同步状态可供其他线程获取；
2.当返回值等于0时，表示获取同步状态成功，但没有可用同步状态了；
3.当返回值小于0时，表示获取同步状态失败。

public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
 //返回值小于0，获取同步状态失败，排队去；
 //获取同步状态成功，直接返回去干自己的事儿。
            doAcquireShared(arg);
}

private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        //构造一个共享结点，添加到同步队列尾部。
        //若队列初始为空，先添加一个无意义的傀儡结点，再将新节点添加到队列尾部。
        boolean failed = true;//是否获取成功
        try {
            boolean interrupted = false;//线程parking过程中是否被中断过
            for (;;) {//死循环
                final Node p = node.predecessor();//找到前驱结点
                if (p == head) {
                //头结点持有同步状态，只有前驱是头结点，才有机会尝试获取同步状态
                    int r = tryAcquireShared(arg);//尝试获取同步装填
                    //r>=0,获取成功，r 值表示剩余可用同步状态
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
//获取成功就将当前结点设置为头结点，若还有可用资源，
//传播下去，也就是继续唤醒后继结点
                        p.next = null; // 方便GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                //是否能安心进入parking状态
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())//阻塞线程
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

//Sets head of queue, and checks if successor may be waiting
// * in shared mode, if so propagating if either propagate > 0 or
//* PROPAGATE status was set.
 private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
         /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
        if (propagate > 0 || h == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

 public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();//释放同步状态
            return true;
        }
        return false;
    }

 private void doReleaseShared() {
        for (;;) {
 //死循环，共享模式，持有同步状态的线程可能有多个，采用循环CAS保证线程安全
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;          
                    unparkSuccessor(h);//唤醒后继结点
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                
            }
            if (h == head)              
                break;
        }
    }