java并发编程学习之AQS

原理

全称AbstractQueuedSynchronizer，当线程去获取资源的时候，会根据状态值state来判断是否有锁，如果有锁，则加入到链表，链表里的线程，通过自旋，判断资源是否已经释放，如果释放，则获取资源。

AQS结构

1. volatile Node head：阻塞的头节点
2. volatile Node tail：阻塞的尾节点，新的阻塞节点加到最后
3. volatile int state：锁的状态，0说明未占用，大于等于1说明已占用
4. Thread exclusiveOwnerThread：占用锁的当前线程

node：双向链表的节点信息

1. Node EXCLUSIVE：独占模式
2. volatile Thread thread：当前线程
3. volatile Node prev：前置节点
4. volatile Node next：后置节点
5. volatile int waitStatus：状态字段，-1：等待被唤醒，大于0，被取消

源码分析

这里以非公平锁为例

加锁

lock

获取锁

final void lock() {
    if (compareAndSetState(0, 1))//如果状态是0，则设置为1
        setExclusiveOwnerThread(Thread.currentThread());//设置占用锁为当前的线程
    else
        acquire(1);//资源被占用
}

acquire

锁被占用后，再尝试获取，获取不到，进入阻塞队列

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

tryAcquire：

调用的是Sync的nonfairTryAcquire方法。

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
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

addWaiter

把线程封装成node，加入队列

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;//如果cas操作成功，设置双向链表
            return node;
        }
    }
    enq(node);
    return node;
}

enq

通过自旋的方式，加入到尾节点

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;//获取尾节点
        if (t == null) { // 如果尾节点为空，初始化头节点和尾节点，地址为同一个
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {//加入尾节点
                t.next = node;
                return t;
            }
        }
    }
}

acquireQueued

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)) {//如果前置节点是head，并且获取到了锁
                setHead(node);//把当前节点设置到head上面
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //如果既不是队头，或者没有抢过其他线程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())//如果队头是唤醒的状态，就用parkAndCheckInterrupt挂起
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

shouldParkAfterFailedAcquire

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;//唤醒，返回true
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;//如果被取消了，被取消的前置节点替换当前节点的前置节点
        } while (pred.waitStatus > 0);
        pred.next = node;//双向链表
    } else {
        /*
         * waitStatus must be 0 or 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;
}

唤醒

unlock

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

release

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);//唤醒头节点
        return true;
    }
    return false;
}

tryRelease

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;//可能重入的情况
    if (Thread.currentThread() != getExclusiveOwnerThread())//非当前线程抛异常
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {//不用cas是因为仅有当前显示有锁
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

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);//如果头节点当前waitStatus<0, 修改为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;
    //唤醒下一个节点，如果第一个为空，从尾部遍历上去，获取最前面的waitStatus 小于0的节点
    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);//唤醒
}