并发编程学习笔记(5)----AbstractQueuedSynchronizer（AQS）原理及使用

（一）什么是AQS?

阅读java文档可以知道，AbstractQueuedSynchronizer是实现依赖于先进先出 (FIFO) 等待队列的阻塞锁和相关同步器（信号量、事件，等等）提供一个框架，它是一个依靠单个原子 int 值来表示状态的大多数同步器的一个基础类。在jdk中他的实现的类有Semaphore，ReentrantLock，CountDownLatch，ReentrantReadWriteLock等等很多的实现。

（二）原理

　　它通过实现一个volatile int state来维护线程的状态，并使用一个双向链表来维护多个线程的等待队列。一般将子类作为一个非公共的内部帮助器类。它存在有独享和共享两种模式，在独享模式下，当锁被占用时，其他线程试图获取锁一定不会成功，共享模式下其他线程获取锁可能会成功。（读写锁读-读不互斥，读-写，写-写互斥）

它提供了getState()、setState(int) ,compareAndSetState(int, int) 三个方法来获取和修改单个原子状态的方法，在使用的是我，我们只需要重写以下几个方法即可，

• tryAcquire(int) (独享模式锁)
• tryRelease(int) （释放独享模式锁）
• tryAcquireShared(int)（共享模式锁）
• tryReleaseShared(int) （释放共享模式所）
• isHeldExclusively() （判断是否是独享模式锁）

（三）源码实现

acquire：　　

　　根据调用流程acquire-release来一步步分析独享模式的源码，共享模式会在后面的学习中补上，首先从acquire,锁的入口开始。

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

　　1)tryAcquire（int）最少会执行一次尝试去获取资源，如果获取到锁，则返回true,否则放回false。

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

tryAcquire（int）在AQS源码中直接直接抛出了一个异常，这是因为我们前面说过AQS是一个框架，使用的时候是需要我们去实现自己的核心部分的，tryAcquire（int）这里调用的实际上是在使用时我们重写的方法，通过操作state
来进行我们自己的实现的。

　　2）addWaiter(Node.EXCLUSIVE)将一个独占模式的线程加入到队列中队尾。

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;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

这里的Node包含了当前线程本身以及线程的状态及以下基本信息，这里就直接看注释，我就不一一写出来了

/** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
        static final int PROPAGATE = -3;

        /**
         * Status field, taking on only the values:
         *   SIGNAL:   值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。  
         *   CANCELLED: 值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化
         *   CONDITION: 值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁
　　　　　 *   PROPAGATE:  值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。

　　这里特别说下一下0状态表示为初始化状态。

　　addWaiter()方法创建一个当前线程节点，当尾节点不为空时，将当前节点的prev指向双向链表的尾节点，并将当前节点通过cas的方式设置为尾节点，将尾节点的next指向当前节点，成功将整个链表连起来，这就是将当前线程加入到队列的过程，返回当前创建的节点。当尾节点为空时，调用enq方法。

　　enq(node)方法

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

　　for()循环表示自旋cas，知道成功将节点添加到队列为止，如果此时的尾节点tail为空，则创建一个空标志性的节点作为head节点，并将尾节点t也指向它，当tail不为空时，就按照前面的正常的添加节点的方式将当前节点添加到队列尾部。添加成功，退出循环，返回当前节点。

　　4）acquireQueued（）使线程在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。

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

　　当线程获取资源失败，并且已经进入了队列中了，那么此时的线程就需要等待了，只等待其他线程执行完成之后，再唤醒队列中的线程时，唤醒到该线程时，该线程才可以继续执行，failed标记线程是否拿到锁，interrupted标记线程是否被中断过。自旋获取锁，

　　第一步，拿到当前的上一个节点，即前驱，如果上一个节点是头节点，则有资格去获取锁，可能是上一个线程释放锁之后轮到第二个线程，当有资格获取锁时，将当前节点设置成头节点，此时前面的一个节点已经出了队列了，并将前一个节点置为空，为了方便垃圾回收，返回等待过程中该线程是否被中断过。

　　shouldParkAfterFailedAcquire（）这个方法，我们也需要理解一下：该方法主要是用于检查状态，判读自己是否可以进入等待队列。

 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
　　
       
        if (ws > 0) {
            /*
             * 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;
    }

　　从方法可以看到，当前节点的前一个节点的状态为通知状态时，返回true,即表示已经告诉了前节点，当释放之后叫醒当前节点，当ws大于0时，表示当前节点的线程被中断或者是等待超时，此时需要将该节点从队列中去除，即node.prev = pred = pred.prev和pred.next=node，表示将pred节点删除，替换成了pred的prev节点;else 则将pred节点设置成为通知唤醒状态。让前驱在执行完成后通知当前节点。

parkAndCheckInterrupt（）使线程真正的进入等待状态。

 private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

　　park()会让当前线程进入waiting状态。在此状态下，有两种途径可以唤醒该线程：1）被unpark()；2）被interrupt()，。需要注意的是，Thread.interrupted()会清除当前线程的中断标记位。

看了shouldParkAfterFailedAcquire()和parkAndCheckInterrupt()，现在让我们再回到acquireQueued()，总结下该函数的具体流程：

结点进入队尾后，检查状态，找到安全休息点；调用park()进入waiting状态，等待unpark()或interrupt()唤醒自己；被唤醒后，看自己是不是有资格能拿到号。如果拿到，head指向当前结点，并返回从入队到拿到号的整个过程中是否被中断过；如果没拿到，继续流程1。

release：
接下来看当线程释放锁，即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;
    }

release的实现就相对简单了，首先同样是调用我们用户自己实现的tryRelease()方法，当tryReleace返回true时，判断头节点状态是否等于初始值，不等于初始值时调用unparkSuccessor(node)方法释放锁，否则表示资源已经呗释放，直接返回true。

unparkSuccessor（node）:

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

这个方法是用于唤醒一个线程，如果这个线程存在的话，ws小于0,表示当前线程是可以呗唤醒的，处于等待状态的线程，此时修改当前线程的状态为原始状态，如果当前线程的下一个线程为空，或者是状态大于0，表示处于CANCELLED状态，此时循环链表，找到最后一个状态小于0的线程，将找到的线程赋值给s,最后如果成功找到s，则使用unpark唤醒s节点的线程，这样就实现了release,

总结：release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源。

（四） 重写tryAcquire(int) 和 tryRelease(int) 方法实现一个简易的可重入锁

这里直接使用java文档的示例代码：

 class Mutex implements Lock, java.io.Serializable {

    // Our internal helper class
    private static class Sync extends AbstractQueuedSynchronizer {
      // Report whether in locked state
      protected boolean isHeldExclusively() { 
        return getState() == 1; 
      }

      // Acquire the lock if state is zero
      public boolean tryAcquire(int acquires) {
        assert acquires == 1; // Otherwise unused
       if (compareAndSetState(0, 1)) {
         setExclusiveOwnerThread(Thread.currentThread());
         return true;
       }
       return false;
      }

      // Release 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;
      }
       
      // Provide a Condition
      Condition newCondition() { return new ConditionObject(); }

      // Deserialize 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();

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

由于官网的文档实现的是不可重入的锁，所以本人也自己实现了一个可重入锁，下面是代码：

package com.ucar.work;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @since: 2018/9/27
 * @version: 1.0
 * Copyright: Copyright (c) 2018
 */
public class MyLock implements Lock, java.io.Serializable {

    private static class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        @Override
        public boolean tryAcquire(int acquires) {
           int stat = getState();
           Thread thread = Thread.currentThread();
           //表示线程第一次进如，直接加锁
           if (stat == 0) {

               compareAndSetState(0,stat+acquires);
               setExclusiveOwnerThread(thread);
               return true;
               //当前线程等于getExclusiveOwnerThread()表示线程重入。返回true表示可以获取锁
           } else if (stat != 0  && thread == getExclusiveOwnerThread()){
               return true;
           }
           //否则返回false表示获取锁失败
           return false;
        }

        @Override
        protected boolean tryRelease(int releases) {
            int stat = getState();
            //stat==0表示该锁状态为释放状态，不能去释放
            if (stat == 0) {
                setExclusiveOwnerThread(null);
                return false;
            } else {
                //重入的情况下减去重入次数
                setState(getState()-releases);
            }
            //返回可释放信号
            return true;
        }

        Condition newCondition() { return new ConditionObject(); }

        private void readObject(ObjectInputStream s)
                throws IOException, ClassNotFoundException {
            s.defaultReadObject();
            setState(0);
        }
    }

    private final Sync sync = new Sync();

    @Override
    public void lock()                { sync.acquire(1); }
    @Override
    public boolean tryLock()          { return sync.tryAcquire(1); }
    @Override
    public void unlock()              { sync.release(1); }
    @Override
    public Condition newCondition()   { return sync.newCondition(); }
    public boolean isLocked()         { return sync.isHeldExclusively(); }
    public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
    @Override
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    @Override
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
}

实现细节注释已经写清楚了，这里就不在叙述了。

至于共享模式的AQS，会在以后的学习中继续记录，不足之处，忘各位大佬多多指点。

原文  并发编程学习笔记(5)----AbstractQueuedSynchronizer（AQS）原理及使用