多线程基础(十四)：AbstractQueuedSynchronizer源码分析

Acquire:
    while (!tryAcquire(arg)) {
       将线程入队
       阻塞当前线程
    }
 Release:
    if (tryRelease(arg))
        取消阻塞第一个排队的线程

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)) {
      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 llegalMonitorStateException();
   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();

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

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

/**
 * Head of the wait queue, lazily initialized.  Except for
 * initialization, it is modified only via method setHead.  Note:
 * If head exists, its waitStatus is guaranteed not to be
 * CANCELLED.
 */
private transient volatile Node head;

/**
 * Tail of the wait queue, lazily initialized.  Modified only via
 * method enq to add new wait node.
 */
private transient volatile Node tail;

/**
 * The synchronization state.
 */
private volatile int state;

/**
 * Returns the current value of synchronization state.
 * This operation has memory semantics of a {@code volatile} read.
 * @return current state value
 */
 
 /**
 * The number of nanoseconds for which it is faster to spin
 * rather than to use timed park. A rough estimate suffices
 * to improve responsiveness with very short timeouts.
 */
static final long spinForTimeoutThreshold = 1000L;

protected AbstractQueuedSynchronizer() { }

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

/** 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;

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

/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;

public final void acquire(int arg) {
    //如果调用tryAcquire方法不成功且acquireQueued入队不成功，则调用自我中断。
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

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

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 (;;) {
            //拿到p的前一个节点
            final Node p = node.predecessor();
            //如果p不为head
            if (p == head) {
                //则说明p没有获得锁，则p继续调用try方法
                int r = tryAcquireShared(arg);
                //如果try方法返回结果大于0，则说明互获得锁 
                if (r >= 0) {
                    //将当前节点移除
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    //如果出现异常，则自我中断
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            //如果p为head节点，则调用shouldParkAfterFailedAcquire对节点状态和指针进行check。如果通过，则将线程park，之后等待被唤醒，被唤醒之后调用中断方法。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        //如果失败，则调用取消入队方法
        if (failed)
            cancelAcquire(node);
    }
}

public final void acquireInterruptibly(int arg)
        throws InterruptedException {
        //判断中断状态
    if (Thread.interrupted())
        throw new InterruptedException();
        //调用tryAcquire方法
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg);
}

private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    //创建一个独占类型的节点。
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        //死循环
        for (;;) {
            //拿到节点的上一个节点
            final Node p = node.predecessor();
            //如果为head且调用tryAcquire方法成功，则将当前节点设置为head，将前一个节点移除。
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            //执行shouldParkAfterFailedAcquire检查。
            if (shouldParkAfterFailedAcquire(p, node) &&
               //执行park并检查中断
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
       //如果状态为false，则取消
        if (failed)
            cancelAcquire(node);
    }
}

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();
            //如果前节点为head,则调用tryAcquireShared
            if (p == head) {
                int r = tryAcquireShared(arg);
                //如果tryAcquireShared结果大于0则说明获取锁成功，将前节点清除
                if (r >= 0) {
                    //设置当前节点为head，且调用release方法
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            //调用检查方法
            if (shouldParkAfterFailedAcquire(p, node) &&
                //执行park和中断 
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
       //如果失败则取消入队
        if (failed)
            cancelAcquire(node);
    }
}

private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    //判断超时时间的有效性
    if (nanosTimeout <= 0L)
        return false;
    //根据超时时间计算deadline
    final long deadline = System.nanoTime() + nanosTimeout;
    //添加Node
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
       //死循环
        for (;;) {
            //判断上一个节点是否为head，且是否获得锁
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            //判断是否已经超时
            nanosTimeout = deadline - System.nanoTime();
            //如果超时则返回false 退出
            if (nanosTimeout <= 0L)
                return false;
            //反之则调用参数检查
            if (shouldParkAfterFailedAcquire(p, node) &&
                //如果此使超时时间大于1000纳秒
                nanosTimeout > spinForTimeoutThreshold)
                //调用带超时参数的park方法
                LockSupport.parkNanos(this, nanosTimeout);
            //如果中断则抛出中断异常
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
       //如果失败则取消入队
        if (failed)
            cancelAcquire(node);
    }
}

private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    //判断超时时间
    if (nanosTimeout <= 0L)
        return false;
    //计算deadline
    final long deadline = System.nanoTime() + nanosTimeout;
    //此处添加共享模式的节点
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
       //死循环
        for (;;) {
            //获得前节点
            final Node p = node.predecessor();
            //判断前节点是否为head
            if (p == head) {
                //用共享模式再次获取锁
                int r = tryAcquireShared(arg);
                //如果成功
                if (r >= 0) {
                   //设置head状态并唤醒
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                //执行带超时的park方法
                LockSupport.parkNanos(this, nanosTimeout);
            //线程中断
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        //如果失败则取消
        if (failed)
            cancelAcquire(node);
    }
}

public final boolean release(int arg) {
    //调用tryRelease方法
    if (tryRelease(arg)) {
        //定义head
        Node h = head;
        //如果head不为空，且等待状态不为0
        if (h != null && h.waitStatus != 0)
            //将后续的节点都执行unpark
            unparkSuccessor(h);
        //之后返回true
        return true;
    }
    //如果释放不成功则返回false
    return false;
}

public final boolean releaseShared(int arg) {
   //调用try方法
    if (tryReleaseShared(arg)) {
       //调用do方法
        doReleaseShared();
        return true;
    }
    return false;
}

private void unparkSuccessor(Node node) {
   //判断状态如果为负数，则清理，将其改为0
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    //s为后续节点
    Node s = node.next;
    //如果s不为空，且s状态大于0
    if (s == null || s.waitStatus > 0) {
        s = null;
        //循环遍历
        for (Node t = tail; t != null && t != node; t = t.prev)
            //如果t的状态小于0 将s赋值为t 
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        //将s对应的线程唤醒
        LockSupport.unpark(s.thread);
}

 private void doReleaseShared() {
//死循环
for (;;) {
    Node h = head;
    //如果h不为空
    if (h != null && h != tail) {
        int ws = h.waitStatus;
        //获得h的状态为ws，如果为SIGNAL
        if (ws == Node.SIGNAL) {
            //将其改为0 如果修改失败则ccontinue 下次循环继续修改
            if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                continue;            // loop to recheck cases
            //将后续的node调用unpark
            unparkSuccessor(h);
        }
        //反之 如果ws状态不为SIGNAL 则 将其从0改为PROPAGATE状态 如果失败则continue
        else if (ws == 0 &&
                 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
            continue;                // loop on failed CAS
    }
    //如果h为head 则直接break结束循环
    if (h == head)                   // loop if head changed
        break;
}
}