AbstractQueuedSynchronizer源码阅读(1)(AQS JDK1.8)

前言

AbstractQueuedSynchronizer（1）(JDK 1.8)

用途

Provides a framework for implementing blocking locks and related
synchronizers (semaphores, events, etc) that rely on
first-in-first-out (FIFO) wait queues.

我的理解：Java并发包里提供的一个用于实现锁和一系列同步器的一个框架。

主要源码分析

本身是一个抽象类，并发包里锁和同步器的实现大都继承了它。

Node内部类

• Node类是一个静态final的内部类，主要用于实现一个等待队列。
• 有独占模式和共享模式两种。
• 主要的属性如下，其中比较重要的就是这个waitStatus，值为如下时的含义:
  1. SIGNAL(-1): 在当前节点释放或被取消时，唤醒后继节点；
  2. CANCELLED(1): 该节点不会再被阻塞,会被移除等待队列；
  3. CONDITION(-2):该节点当前处于一个条件等待队列里(应该给lock的Condition有关)；
  4. PROPAGATE(-3):用于共享模式，暂时不清楚；
  5. 0：正常在队列中，如果是实现非公平锁，应该会拥有竞争锁的权利；
• next和nextWaiter的区别，主要就是nextWaiter用于表示下一个处于wait condition队列的节点或共享模式的节点，next表示同步队列中下一个可以唤醒的节点；

volatile int waitStatus;

volatile Node prev;

volatile Node next;

volatile Thread thread;

Node nextWaiter;

ps： 如果处于独占模式下，该等待队列应该会逻辑上维护有两种队列，一个是可以正常竞争锁(如果用于实现锁的话)以及释放唤醒的节点同步队列，一个是处于等待某种条件的节点队列。

ConditionObject类

主要用于实现前面说的处于等待某种条件的节点队列，实现了相关入队出队操作，同时实现了Condition接口，其接口如下：

public interface Condition {

    void await() throws InterruptedException;

    void awaitUninterruptibly();
    
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    
    boolean awaitUntil(Date deadline) throws InterruptedException;

    void signal();
    
    void signalAll();
}

ps: 底层实现原子操作的，用到了unsafe包CAS操作。

重要方法

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
  	// 添加到条件等待队列中
    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;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

public final void signal() {
  	// 判断当前锁是否被当前线程独占，如果不是则抛出IllegalMonitorStateException
    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);
}

主要的属性及方法

主要属性

private transient volatile Node head;

private transient volatile Node tail;

/**
 * The synchronization state.
 */
private volatile int state;	// 在锁实现里表示的是持有该锁的线程数

private static final Unsafe unsafe = Unsafe.getUnsafe();

重要方法

1. acquire(int arg)

/**
 * 独占模式，
 * 先尝试获取锁，失败则入Condition wait队列中
 * 忽略Interrupt
 */
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

// 入队默认waitStatus是0
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;
}

// 入队后再次尝试获取锁
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;
            }
            // 获锁失败后判断是否应该进行阻塞 使用LockSupport.park(this);底层还是用的unsafe包
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        // 获锁失败后
        if (failed)
            cancelAcquire(node);
    }
}

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

主要要实现的方法

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

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

protected int tryAcquireShared(int arg) {
  	throw new UnsupportedOperationException();
}

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

protected boolean isHeldExclusively() {
  	throw new UnsupportedOperationException();
}