ReentrantLock 与 AQS 源码分析

public interface Lock {
    // 阻塞获取锁，如果获取不到锁就一直等待
    void lock();
    // 可中断获取锁，在获取锁的过程可以被中断，但是 Synchronized 是不可以
    void lockInterruptibly() throws InterruptedException;
    // 非阻塞获取锁，没有获取到锁立即返回
    boolean tryLock();
    // 超时获取锁，没获取到锁等待一段时间
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    // 解锁
    void unlock();
    // 等待唤醒机制的条件
    Condition newCondition();
}

// 同步锁
private final Sync sync;

//同步队列头指针
private transient volatile Node head;
// 同步队列尾指针
private transient volatile Node tail;
// 状态标志，0 则没有线程在临界区，非零表示有 state 个线程在临界区（由于锁可重入）
private volatile int state;

static final class Node {
        // 共享非 null/独占为 null  
        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;
       // 双向链表  这两个指针用于同步队列构建链表使用的   下面还有一个 nextWaiter 是用来构建等待单链表队列
        volatile Node prev;
        volatile Node next;
        // 线程
        volatile Thread thread;
        // 等待队列单链表
        Node nextWaiter;
        /**
         * Returns true if node is waiting in shared mode.
         */
        final boolean isShared() {
            return nextWaiter == SHARED;
        }
    }

public final void await() throws InterruptedException {
            // 等待可中断
            if (Thread.interrupted())
                throw new InterruptedException();
            // 加入等待队列， new 新的 Node 做一个尾插入
            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;
            // 清理队列中状态不是 Condition 的的任务，包括被唤醒的 SIGNAL 和 被取消的 CANCELLED
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            //被中断 抛异常
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

public final void signal() {
    // 获取独占锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 唤醒等待队里中的第一个线程
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

private void doSignal(Node first) {
           do {
               // 头指针指向唤醒节点的下一个节点，并顺便判断等待队列是否空
               if ( (firstWaiter = first.nextWaiter) == null)
                   lastWaiter = null;
               // 解除引用
               first.nextWaiter = null;
           } while (!transferForSignal(first) && (first = firstWaiter) != null); //移入同步队列失败则继续唤醒下一个线程，否则唤醒成功
           // 唤醒成功的线程不一定马上能开始执行，只有在前驱节点被取消或者没办法被唤醒时
  }
  
  
  //  将节点从等待队列移动到同步队列   成功返回 true 失败 false
   final boolean transferForSignal(Node node) {
       // 在等待队列中的节点只有 condition 和 cancelled 两种状态，如果状态更新失败说明任务被取消
       // 否则更新为初始状态   直接返回的话上面的 doSignal 就会继续唤醒后面的线程
       if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
           return false;
       // 把当前节点加入同步队列
       Node p = enq(node);
       // 获取同步队列中倒数第二个节点的状态，当前节点的前驱
       int ws = p.waitStatus;
       // 如果前驱节点被取消或者在设置前驱节点状态为Node.SIGNAL状态失败时，唤醒被通知节点代表的线程
       if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
           LockSupport.unpark(node.thread);
       return true;
   }
   
   
   // 插入一个节点到同步队列，如果同步队列是空的则加入一个空节点做为头结点
   // 死循环保证肯定能插入    返回插入节点的前驱
   private Node enq(final Node node) {
       for (;;) {
           Node t = tail;
           if (t == null) { // Must initialize
               if (compareAndSetHead(new Node()))
                   tail = head;
           } else {
               // 这一步不需要 cas 是因为并发没关系，只是指向链表结尾，不会多线程更新问题
               node.prev = t;
               // 可能有多个线程抢
               if (compareAndSetTail(t, node)) {
                   t.next = node;
                   return t;
               }
           }
       }
   }

package util.AQSTest;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
// test signal 执行后不会导致当前线程立即释放锁
public class AQSTest {
    static Lock lock = new ReentrantLock();
   static Condition run1Cond = lock.newCondition();
    static Condition run2Cond = lock.newCondition();
    static class Runner1 implements Runnable {
        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println("runner 1 start");
                run1Cond.await(1, TimeUnit.SECONDS);
                run2Cond.signal();
                System.out.println("runner 1 exit");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }
    static class Runner2 implements Runnable {
        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println("runner 2 start");
                run2Cond.await();
                System.out.println("runner 2 exit");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
        }
    }
    public static void main(String[] args) {
        new Thread(new Runner1(),"runner1").start();
        new Thread(new Runner2(),"runner2").start();
    }
}

runner 1 start
runner 2 start
runner 1 exit
runner 2 exit

// 释放锁的公共操作
protected final boolean tryRelease(int releases) {
    // 释放锁首先就是使用 AQS 中的 state 的值减去信号量 判断是否为0
    // 如果是 0 则表明成功释放锁，独占线程设为 null，否则说明还占用锁
    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;
}

    公平锁执行 lock 操作就是执行了 AQS 中的 acquire(1) 也就是请求一个锁资源。但是注意，在 AQS 中的 acquire 中的 tryAcquire 方法没有实现，所以必须由当前类实现。

    在 tryAcquire 中做的事情就是看是否有代码在临界区。没有则还要看同步队列中是否有线程等待，当只有这一个线程在获取锁的时候才能正常的获取锁，其他情况都失败。

// 公平锁
    static final class FairSync extends Sync {
        final void lock() {
            acquire(1);
        }
        // 没有代码在临界区或者是当前线程的重入 则获取成功，否则失败
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            // 如果当前线程在获取锁的过程没有其他线程在临界区
            if (c == 0) {
                // 如果同步队列中没有等待的线程，就设置 state ，并且当前线程设为独占线程
                if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 有程序在临界区，如果是当前线程可重入，加上请求的资源数
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            // 竞争锁失败，因为他是公平的锁竞争
            return false;
        }
    }

// 非公平锁
static final class NonfairSync extends Sync {
    final void lock() {
        // 如果没有代码在临界区 直接获取锁，独占
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
        // 有代码在临界区则执行尝试获取锁
            acquire(1);
    }
    // 和公平锁中的 tryAcquire 一模一样只是少了关于同步队列中是否有等待线程的判断
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}
final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        // 没有线程获取锁 直接获取到锁  和公平锁中的 tryAcquire 一模一样只是少了关于同步队列的判断
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        // 重入锁
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

// 对状态变量的 volatile 读写
protected final int getState() {
    return state;
}
protected final void setState(int newState) {
    state = newState;
}

//  尝试释放锁
   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) {
       // node 一般就是当前正在运行的线程
       int ws = node.waitStatus;
       // 当前线程置为初始状态   可以失败
       if (ws < 0)
           compareAndSetWaitStatus(node, ws, 0);
       // 找到同步队列中的下一个节点
       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 int fullyRelease(Node node) {
       boolean failed = true;
       try {
           int savedState = getState();
           if (release(savedState)) {
               failed = false;
               return savedState;
           } else {
               throw new IllegalMonitorStateException();
           }
       } finally {
           if (failed)
               node.waitStatus = Node.CANCELLED;
       }
   }

// 获取锁
    public final void acquire(int arg) {
        // 尝试获取锁 失败则加入同步队列 如果是同步队列中的第一个线程就自旋获取锁
        // 上面的步骤的自旋获取锁阶段，返回的是是否需要中断，所以下面就进行 selfInterrupt
        // tryAcquire 是模板方法，因为对于公平锁和非公平锁获取锁方式不同
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    
    
    // 创建一个节点放入到同步对列中   可传入是否为独占锁   返回当前节点
    private Node addWaiter(Node mode) {
        // 默认的 status 是 0
        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;
            // 把 tail 设置为 node 成功说明没有竞争
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 失败则就说明空队列   创建头结点
        enq(node);
        return node;
    }
    
     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)) {
                    // 把当前节点设置为头结点  thread=null 也就可以看做当前线程在运行，所以就不在同步队列
                    setHead(node);
                    // gc
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // 如果获取锁失败，检测为 SIGNAL 或者设置为 SIGNAL 然后让此线程等待 等待操作在 parkAndCheckInterrupt 中完成
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            // 失败 取消
            if (failed)
                cancelAcquire(node);
        }
    }