深入学习java并发编程：Lock与AbstractQueuedSynchronizer（AQS）实现

1、相关类图



2、Lock接口
       锁是用来控制多个线程访问共享资源的方式，一般来说，一个锁能防止多个线程同时访问共享资源（但是有些锁可以允许多个线程并发的访问共享资源，比如读写锁）。在Lock接口出现之前，Java程序是依靠synchronized关键字实现锁的功能的，而从java SE5之后，并发包中新增了Lock接口（以及相关实现类）用来实现锁的功能，只是在使用是需要显式的获取和释放锁，虽然他缺少了隐式获取释放锁的便捷性，但是却拥有了锁获取与释放的可操作性、可中断性以及超时获取锁等多种synchronized关键字所不具备的同步特性。
      Lock的使用很简单：
 
              Lock l = new RentrantLock() ;
                 l.lock();   //不要放在try块中
                   try {
                              // access the resource protected by this lock
                      } finally {
                            l.unlock();
                     }
  在finally中释放锁，目的是保证在获取到锁后，最终释放锁。
        Lock接口提供的锁，所具有的synchronized关建字不具备的特性：

     1）尝试非阻塞的获取锁：当前线程尝试获取锁，如果没被占用，则成功并持有锁
     2）能被中断的获取锁：与synchronized不同，获取到锁的线程能够响应中断，当获取到锁的线程被中断时，中断异常将会抛出，同时锁会被释放
     3）超时获取锁：在指定时间内获取锁，如果时间到了，仍旧无法获取到锁，则返回

3、AbstractQueuedSynchronizer抽象队列同步器AQS
       队列同步器，是用来构建锁或者其他同步组件的基础架构，它使用一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。同步器的主要使用方式是继承，子类通过继承同步器并实现他的抽象方法来管理同步状态，在抽象方法实现过程中免不了对同步状态进行修改，这时使用同步器提供的3个方法getState（）、setState(int newState）和compareAndSet（int
expect ， int update）来进行操作，他们保证了状态的改变是线程安全的。子类推荐被定义为同步组件的自定义内部类，同步器自身没有实现任何同步接口，仅仅定义了若干同步状态的获取和释放的方法供自定义同步组件使用，同步器既可以支持独占式获取同步状态，也可以支持共享式获取同步状态。
        同步器是实现锁以及其他同步组件的关键，在锁的实现中聚合同步器，利用同步器实现锁的语义。二者的关系是：锁面向使用者，同步器面向锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程排队、等待与唤醒等底层操作。
1）队列同步器的接口和重要方法
       同步器的设计是基于模板方法模式，使用者需要继承同步器并重写指定的方法，然后将同步器组合到自定义的同步组件中，并调用同步器提供的模板方法，而这些模板方法会调用使用者重写的方法。
      获取以及修改同步状态的方法：
           public  int  getState（）：获取当前同步状态
           public    void  setState（int newState）：设置当前同步状态为newState
           public    boolean compareAndSet（int expect ，int update）：使用 CAS设置当前状态，该方法保证修改状态的原子性 
<
4000
div style="font-family:'微软雅黑';font-size:14px;line-height:21px;">     同步器可重写的方法：
            protected boolean tryAcquire(int arg)    ：独占式的获取同步状态，实现该方法需要查询同步状态并判断是否符合预期，然后在进行CAS                  
                                                                                   设置同步状态
            protected boolean tryRelease(int arg)   ：独占式释放同步状态，等待获取同步状态的线程将有机会获取同步状态
            protected int tryAcquireShared(int arg)  ：共享式获取同步状态，，返回大于等于0的值，表示获取成功，反之，获取失败
            protected boolean tryReleaseShared(int arg) ：共享式释放同步状态
            protected boolean isHeldExclusively()  :   当前同步器是否在独占模式下被线程占用，一般该方法表示是否被当前线程所独占
   同步器提供的模板方法： 自定义组件时将会调用这些方法，这些方法将会调用自定义重写的相关方法
            public final void acquire(int arg) ：独占式的获取同步状态，如果获取成功，则返回，否则，将会进入同步队列等待，该方法将会调用重写
                                                                         的tryAcquire（int arg）方法
            public final void acquireInterruptibly(int arg)：与acquire(int arg)相同，但是该方法响应中断，若当前线程被中断，则抛出中断异常
            public final boolean tryAcquireNanos(int arg, long nanosTimeout)：在acquireInterruptibly(int arg)的基础上增加了超时限制，如超
                                                                               时没有获得同步状态，则返回false，否则返回true
            public final boolean release(int arg)：独占式的释放状态，该方法会在释放状态后，将同步队列中第一个节点包含的线程唤醒
            public final void acquireShared(int arg)：共享式的获取同步状态，实现该方法需要查询同步状态并判断是否符合预期，然后在进行CAS                  
                                                         设置同步状态，如果未获取到状态将进入同步队列等待，与独占式获取不同的是可以有多个线程获取到同步状态
            public final void acquireSharedInterruptibly(int arg)：
            public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) ：

             public final boolean releaseShared(int arg)：共享式释放同步状态

2）队列同步器的锁应用实例

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

4、AbstractQueuedSynchronizer的实现
       同步器依赖内部的同步队列（FIFO队列）来完成同步状态的管理，当前线程获取同步状态失败后，同步器会将当前线程以及等待状态等信息构造为一个Node节点并将其加入同步队列，同时阻塞线程，当释放状态时，把首节点中的线程唤醒，使其再次尝试获取同步状态。
      Node节点类型：
                 static final class Node {
                                       static final Node SHARED = new Node();
                                        static final Node EXCLUSIVE = null;
                   static final int CANCELLED =  1;
                   static final int SIGNAL    = -1;
                   static final int CONDITION = -2;
                   static final int PROPAGATE = -3;
                             volatile int waitStatus;
                             volatile Node prev;
                             volatile Node next;
                             volatile Thread thread;
                              Node nextWaiter;
              final boolean isShared() {
            return nextWaiter == SHARED;
        }
 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;
        }
 }
    节点是构成同步队列的基础，同步器拥有首节点head和尾节点tail，没有成功获取同步状态的线程将会成为节点加入到队列的尾部，队列的基本结构如图：



    
  独占式同步状态获取与释放：
    通过调用同步器的acquire（int arg）方法可以获取同步状态，这个方法对中断不敏感。方法如下：
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
     该方法主要完成同步状态的获取、节点构造、加入队列以及在同步队列中自旋等待的相关工作，主要逻辑是：首先调用自定义同步器实现的tryAcquire（int arg）方法，该方法保证线程安全的获取同步状态，如果同步状态获取失败，则构造同步节点并通过addWaiter（Node node）方法将该节点加入到同步队列的尾部，最后调用acquireQueued（Node
node ，int arg）方法，使得该节点以“死循环”的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出对或阻塞线程被中断来实现。
     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;
    }
     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;
                }
            }
        }
    }
代码中通过compareAndSetTail方法确保节点能够线程安全的添加。enq中通过"死循环"不断试着添加节点，直到成功。可以看到通过CAS，使得并发添加节点串行化。
     节点进入同步队列后，就进入了一个自旋的过程，每个节点都在观察，当当前节点的前驱节点是头结点，并且获取到了同步状态，就可以从这个自旋过程退出。当自旋结束，同步状态获取成功，线程就从acquire（int arg）中返回，。至此，线程获取了同步状态，，对锁来说，就是获得了锁。
    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);
        }
    }
当前线程获取同步状态并处理后，就需要释放同步状态，使得后续节点能够继续获取同步状态。通过release（int arg）方法可以释放同步状态，该方法在释放了同步状态后，会唤醒其后继节点（进而使后继节点重新尝试获取同步状态）。代码如下：
  public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    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);
    }
     共享式同步状态获取与释放：
     通过调用同步器的acquireShared（Int arg ）可以共享式的获取同步状态，在该方法中，通过调用自定义的tryAcquireShared（int
arg）获取同步状态，如果返回值大于等于0，就表示获取到同步状态，结束自旋。
   public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
同独占式的释放一样，共享式的释放调用releaseShared(int arg)方法释放同步状态。
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
    private void doReleaseShared() {
        /*
         * Ensure that a release propagates, even if there are other
         * in-progress acquires/releases.  This proceeds in the usual
         * way of trying to unparkSuccessor of head if it needs
         * signal. But if it does not, status is set to PROPAGATE to
         * ensure that upon release, propagation continues.
         * Additionally, we must loop in case a new node is added
         * while we are doing this. Also, unlike other uses of
         * unparkSuccessor, we need to know if CAS to reset status
         * fails, if so rechecking.
         */
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }