Java之AQS框架底层原理分析

//1、下面两个对象分别代编代表共享锁和独占锁(排它锁)，可以理解成枚举
static final Node SHARED = new Node();//共享锁
static final Node EXCLUSIVE = null;//排它锁

//2、四种等待状态
/** 表示线程获取锁的请求已经取消了 */
static final int CANCELLED =  1;
/** 下一个节点(线程)需要被唤醒 */
static final int SIGNAL    = -1;
/** 表示节点在等待队列中，节点线程等待唤醒 */
static final int CONDITION = -2;
/** 共享锁模式下才会用到 */
static final int PROPAGATE = -3;

//3、常见变量
volatile int waitStatus;//等待状态
volatile Node prev;//同步队列-前置节点
volatile Node next;//后置节点
volatile Thread thread;//节点包含的线程-类似线程池的Worker持有线程
Node nextWaiter;//指向下一个处于CONDITION状态的节点/SHARED

2、state变量。

public void lock() {
    sync.lock();
}

final void lock() {

    if (compareAndSetState(0, 1))//1 
        setExclusiveOwnerThread(Thread.currentThread());//2
    else
        acquire(1);
}

final void lock() {
    acquire(1);
}

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {//1-是
        if (compareAndSetState(0, acquires)) {//2
            setExclusiveOwnerThread(current);//2-成功
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {//1-否
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);//3-是
        return true;
    }
    return false;//2-失败、3-否
}

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (!hasQueuedPredecessors() && //1也体现了公平锁的特点-前面有人就去排队
            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;
}

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);//1
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {//2
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);//3
    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)) {//1
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
          	//不是头节点的后驱节点&nbs***bsp;尝试获取锁失败
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);//2
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * waitStatus must be 0&nbs***bsp;PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

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

public void unlock() {
  	//调用AQS的release方法，这个方法是框架提供的
    sync.release(1);
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {//调用实现类的方法，我们看看开发者如何自定义的
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);//唤醒头节点的下一个节点==》LockSupport.unpark(h.next.thread);
        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;
}

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);//1

    Node s = node.next;//2
    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);//3
}

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

ReentrantLock lock = new ReentrantLock();
Condition conditionA = lock.newCondition();
Condition conditionB = lock.newCondition();

public final void wait() throws InterruptedException {
    wait(0);
}

//使用示例
synchronized (obj) {
    while (some conditions)
        obj.wait();
       ... 
}

public final native void notify();

public final void await() throws InterruptedException {
    if (Thread.interrupted())//1
        throw new InterruptedException();
    Node node = addConditionWaiter();//2
    int savedState = fullyRelease(node);//3
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {//4
        LockSupport.park(this);//5
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//6
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//7
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) 
        unlinkCancelledWaiters();//8
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);//9
}

private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}

final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        enq(node);
        return true;
    }
    /*
     * If we lost out to a signal(), then we can't proceed
     * until it finishes its enq().  Cancelling during an
     * incomplete transfer is both rare and transient, so just
     * spin.
     */
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}

public final void signal() {
    if (!isHeldExclusively())//1
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;//2
    if (first != null)//3
        doSignal(first);//4
}

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

/*
 * Transfers a node from a condition queue onto sync queue.
 */
final boolean transferForSignal(Node node) {
  
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))//1
        return false;

    Node p = enq(node);//2
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//3
        LockSupport.unpark(node.thread);
    return true;
}

void Parker::unpark() {
  int s, status ;
 //给互斥量加锁，如果互斥量已经上锁，则阻塞到互斥量被解锁
 //park进入wait时，_mutex会被释放
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant") ; 
  //存储旧的_counter
  s = _counter; 
//许可改为1，每次调用都设置成发放许可
  _counter = 1;
  if (s < 1) {
     //之前没有许可
     if (WorkAroundNPTLTimedWaitHang) {
      //默认执行 ,释放信号，表明条件已经满足，将唤醒等待的线程
        status = pthread_cond_signal (_cond) ;
        assert (status == 0, "invariant") ;
        //释放锁
        status = pthread_mutex_unlock(_mutex);
        assert (status == 0, "invariant") ;
     } else {
        status = pthread_mutex_unlock(_mutex);
        assert (status == 0, "invariant") ;
        status = pthread_cond_signal (_cond) ;
        assert (status == 0, "invariant") ;
     }
  } else {
   //一直有许可，释放掉自己加的锁,有许可park本身就返回了
    pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
  }
}

public static void main(String[] args) {
    Thread t1 = new Thread(() -> {
        int count = 100;
        while (count > 0) {
            LockSupport.park();
            System.out.println(count--);
            Thread.interrupted();//清理标志位
        }
    });

    Thread t2 = new Thread(() -> {
        try {
            while (true) {
                Thread.sleep(100);
                t1.interrupt();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    t1.start();
    t2.start();
}

Thread thread = new Thread(() -> {
    while (!Thread.interrupted()) {
        // do more work.
    }
});
thread.start();

// 一段时间以后
thread.interrupt();