Java并发之Lock原理解析

class Node {
  // 一个标识符，指示节点使用共享模式等待
  static final Node SHARED = new Node();
  // 一个标识符，指示节点使用独占模式等待
  static final Node EXCLUSIVE = null;
  // 等待状态值，表示线程已被取消
  static final int CANCELLED =  1;
  // 等待状态值,表示继任的线程需要取消阻塞
  // 当前节点的后继节点已经通过park阻塞，因此当当前节点释放或取消时必须unpark他的后继节点
  static final int SIGNAL    = -1;
  // 等待状态值,表示线程由于阻塞而等待
  static final int CONDITION = -2;
  // 等待状态值,表示处于共享模式下，下一次的acquireShared无条件的传播
  static final int PROPAGATE = -3;
  // 节点当前等待状态值, 0:None of the above
  volatile int waitStatus;
  // 前驱节点
  volatile Node prev;
  // 后继节点
  volatile Node next;
  // 当前线程
  volatile Thread thread;
  // 指向下一个处于阻塞等待的节点
  Node nextWaiter;
}

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 获取node的前驱节点
            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); // Cancels an ongoing attempt to acquire.
    }
}

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

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) {
        /*
          * 前驱节点终止了. 跳过前驱节点，再尝试
          * 此处删除了无效的线程终止节点
          */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 找到一个前驱节点waitStats不是CANCELLED的并且最靠近head节点的那一个为止
        // 用CAS机制把前驱节点的waitStatus更新为SIGNAL状态
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

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

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

private void doAcquireInterruptibly(int arg) throws InterruptedException {
  final Node node = addWaiter(Node.EXCLUSIVE);
  boolean failed = true;
  try {
      for (;;) {
          final Node p = node.predecessor();
          if (p == head && tryAcquire(arg)) {
              setHead(node);
              p.next = null; // help GC
              failed = false;
              return;
          }
          if (shouldParkAfterFailedAcquire(p, node) &&
              parkAndCheckInterrupt())
              throw new InterruptedException(); // 此处是区别，响应中断
      }
  } finally {
      if (failed)
          cancelAcquire(node);
  }
}

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

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 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

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

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();  // 此处不同的处理
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

abstract static class Sync extends AbstractQueuedSynchronizer {
  /**
    * Performs {@link Lock#lock}. The main reason for subclassing
    * is to allow fast path for nonfair version.
    */
  abstract void lock();

  /**
    * Performs non-fair tryLock.  tryAcquire is implemented in
    * subclasses, but both need nonfair try for trylock method.
    */
  final boolean nonfairTryAcquire(int acquires) {
      final Thread current = Thread.currentThread();
      int c = getState();
      if (c == 0) { // 可以获得锁
          if (compareAndSetState(0, acquires)) {
              setExclusiveOwnerThread(current);
              return true;
          }
      }
      else if (current == getExclusiveOwnerThread()) { // 重入
          int nextc = c + acquires;
          if (nextc < 0) // overflow 锁的最大上限int.max
              throw new Error("Maximum lock count exceeded");
          setState(nextc);
          return true;
      }
      return false;
  }

  protected final boolean tryRelease(int releases) {
      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;
  }
  ...
}

static final class FairSync extends Sync {
  private static final long serialVersionUID = -3000897897090466540L;

  final void lock() {
      acquire(1); // 公平锁，从队列head开始
  }

  /**
    * Fair version of tryAcquire.  Don't grant access unless
    * recursive call or no waiters or is first.
    */
  protected final boolean tryAcquire(int acquires) {
      final Thread current = Thread.currentThread();
      int c = getState();
      if (c == 0) {
          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;
  }
}

/**
    * The number of reentrant read locks held by current thread.
    * Initialized only in constructor and readObject.
    * Removed whenever a thread's read hold count drops to 0.
    */
private transient ThreadLocalHoldCounter readHolds;

/**
    * The hold count of the last thread to successfully acquire
    * readLock. This saves ThreadLocal lookup in the common case
    * where the next thread to release is the last one to
    * acquire. This is non-volatile since it is just used
    * as a heuristic, and would be great for threads to cache.
    *
    * <p>Can outlive the Thread for which it is caching the read
    * hold count, but avoids garbage retention by not retaining a
    * reference to the Thread.
    *
    * <p>Accessed via a benign data race; relies on the memory
    * model's final field and out-of-thin-air guarantees.
    */
private transient HoldCounter cachedHoldCounter;

/**
    * firstReader is the first thread to have acquired the read lock.
    * firstReaderHoldCount is firstReader's hold count.
    *
    * <p>More precisely, firstReader is the unique thread that last
    * changed the shared count from 0 to 1, and has not released the
    * read lock since then; null if there is no such thread.
    *
    * <p>Cannot cause garbage retention unless the thread terminated
    * without relinquishing its read locks, since tryReleaseShared
    * sets it to null.
    *
    * <p>Accessed via a benign data race; relies on the memory
    * model's out-of-thin-air guarantees for references.
    *
    * <p>This allows tracking of read holds for uncontended read
    * locks to be very cheap.
    */
private transient Thread firstReader = null;
private transient int firstReaderHoldCount;

final boolean writerShouldBlock() {
    return false; // writers can always barge
}
final boolean readerShouldBlock() {
    /* As a heuristic to avoid indefinite writer starvation,
        * block if the thread that momentarily appears to be head
        * of queue, if one exists, is a waiting writer.  This is
        * only a probabilistic effect since a new reader will not
        * block if there is a waiting writer behind other enabled
        * readers that have not yet drained from the queue.
        */
    return apparentlyFirstQueuedIsExclusive();   //该方法，具体又是在 AbstractQueuedSynchronizer中
}
//// --> call AbstractQueuedSynchronizer.apparentlyFirstQueuedIsExclusive
/**
     * Returns {@code true} if the apparent first queued thread, if one
     * exists, is waiting in exclusive mode.  If this method returns
     * {@code true}, and the current thread is attempting to acquire in
     * shared mode (that is, this method is invoked from {@link
     * #tryAcquireShared}) then it is guaranteed that the current thread
     * is not the first queued thread.  Used only as a heuristic in
     * ReentrantReadWriteLock.
     */
    final boolean apparentlyFirstQueuedIsExclusive() {
        Node h, s;
        return (h = head) != null &&
            (s = h.next)  != null &&
            !s.isShared()         &&
            s.thread != null;
    }

/**
    * Acquires the read lock.
    *
    * <p>Acquires the read lock if the write lock is not held by
    * another thread and returns immediately.
    *
    * <p>If the write lock is held by another thread then
    * the current thread becomes disabled for thread scheduling
    * purposes and lies dormant until the read lock has been acquired.
    */
public void lock() {
    sync.acquireShared(1);
}
//// --> call AbstractQueuedSynchronizer.acquireShared
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
    // 在 AQS 中，如果 tryAcquireShared(arg) 方法返回值小于 0 代表没有获取到共享锁(读锁)，大于 0 代表获取到
}
//// --> call ReentrantReadWriteLock.Sync.tryAcquireShared
protected final int tryAcquireShared(int unused) {
    /*
        * Walkthrough:
        * 1. If write lock held by another thread, fail.
        * 2. Otherwise, this thread is eligible for
        *    lock wrt state, so ask if it should block
        *    because of queue policy. If not, try
        *    to grant by CASing state and updating count.
        *    Note that step does not check for reentrant
        *    acquires, which is postponed to full version
        *    to avoid having to check hold count in
        *    the more typical non-reentrant case.
        * 3. If step 2 fails either because thread
        *    apparently not eligible or CAS fails or count
        *    saturated, chain to version with full retry loop.
        */
    Thread current = Thread.currentThread();
    int c = getState(); // 获取当前锁状态
    if (exclusiveCount(c) != 0 &&     // 有线程已经抢占了写锁
        getExclusiveOwnerThread() != current)  // 写线程不是当前线程
        return -1;  // 返回-1，进入lock等待队列  
    int r = sharedCount(c); // 获取当前读线程数
    if (!readerShouldBlock() &&  // 当前读线程是否应当排队
        r < MAX_COUNT &&  
        compareAndSetState(c, c + SHARED_UNIT)) { // cas 新增读状态成功
        if (r == 0) { // 如果r=0, 表示，当前线程为第一个获取读锁的线程
            firstReader = current;
            firstReaderHoldCount = 1; // 如果第一个获取读锁的对象为当前对象，将firstReaderHoldCount 加一
        } 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);
}

/**
    * Full version of acquire for reads, that handles CAS misses
    * and reentrant reads not dealt with in tryAcquireShared.
    */
// 第一次尝试获取读锁失败后
final int fullTryAcquireShared(Thread current) {
    /*
        * This code is in part redundant with that in
        * tryAcquireShared but is simpler overall by not
        * complicating tryAcquireShared with interactions between
        * retries and lazily reading hold counts.
        */
    HoldCounter rh = null;
    for (;;) {
        int c = getState();
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
                // 如果当前线程不是写锁的持有者，直接返回-1，
                // 结束尝试获取读锁，需要排队去申请读锁。
            // else we hold the exclusive lock; blocking here
            // would cause deadlock.
        } else if (readerShouldBlock()) {
            // readerShouldBlock() 为 true，说明阻塞队列中有其他线程在等待
            // Make sure we're not acquiring read lock reentrantly
            if (firstReader == current) {
                // firstReader 线程重入读锁，直接到下面的 CAS
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                        // 如果发现 count == 0，也就是说，纯属上一行代码初始化的，那么执行 remove
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        if (compareAndSetState(c, c + SHARED_UNIT)) { // cas获取读锁
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                 // 下面这几行，就是将 cachedHoldCounter 设置为当前线程
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;  // 计数+1
                cachedHoldCounter = rh; // cache for release
            }
            return 1; // 成功获得锁
        }
    }

/**
    * Attempts to release this lock.
    *
    * <p>If the number of readers is now zero then the lock
    * is made available for write lock attempts.
    */
public void unlock() {
    sync.releaseShared(1);
}
//// --> call AbstractQueuedSynchronizer.releaseShared
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared(); // 唤醒后继节点
        return true;
    }
    return false;
}
//// --> call ReentrantReadWriteLock.Sync.tryReleaseShared
protected final boolean tryReleaseShared(int unused) {
    Thread current = Thread.currentThread();
    if (firstReader == current) {
        // assert firstReaderHoldCount > 0;
        if (firstReaderHoldCount == 1) //如果等于 1，那么这次解锁后就不再持有锁了，把 firstReader 置为 null，给后来的线程用
            firstReader = null;
        else
            firstReaderHoldCount--;
    } else {
        // 判断 cachedHoldCounter 是否缓存的是当前线程，不是的话要到 ThreadLocal 中取
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            // 这一步将 ThreadLocal remove 掉，防止内存泄漏。因为已经不再持有读锁了
            readHolds.remove();
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        --rh.count;
    }
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;
        //nextc 是 state 高 16 位减 1 后的值
        if (compareAndSetState(c, nextc))
            // Releasing the read lock has no effect on readers,
            // but it may allow waiting writers to proceed if
            // both read and write locks are now free.
            // 如果 nextc == 0，那就是 state 全部 32 位都为 0，也就是读锁和写锁都空了
            // 此时这里返回 true 的话，其实是帮助唤醒后继节点中的获取写锁的线程
            return nextc == 0;
    }
}

// WriteLock
public void lock() {
    sync.acquire(1);
}
// AQS
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        // 如果 tryAcquire 失败，那么进入到阻塞队列等待
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

// Sync
protected final boolean tryAcquire(int acquires) {

    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c);
    if (c != 0) {

        // 看下这里返回 false 的情况：
        //   c != 0 && w == 0: 写锁可用，但是有线程持有读锁(也可能是自己持有)
        //   c != 0 && w !=0 && current != getExclusiveOwnerThread(): 其他线程持有写锁
        //   也就是说，只要有读锁或写锁被占用，这次就不能获取到写锁
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;

        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");

        // 这里不需要 CAS，仔细看就知道了，能到这里的，只可能是写锁重入，不然在上面的 if 就拦截了
        setState(c + acquires);
        return true;
    }

    // 如果写锁获取不需要 block，那么进行 CAS，成功就代表获取到了写锁
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    setExclusiveOwnerThread(current);
    return true;
}

// unlock
public void unlock() {
    sync.release(1);
}

// AQS
public final boolean release(int arg) {
    // 1. 释放锁
    if (tryRelease(arg)) {
        // 2. 如果独占锁释放"完全"，唤醒后继节点
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

// Sync 
// 释放锁，是线程安全的，因为写锁是独占锁，具有排他性
// 实现很简单，state 减 1 就是了
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);
    // 如果 exclusiveCount(nextc) == 0，也就是说包括重入的，所有的写锁都释放了，
    // 那么返回 true，这样会进行唤醒后继节点的操作。
    return free;
}