图灵学院Java架构师-VIP-并发编程(AQS详解)

1、LockSupport

LockSupport类的核心方法其实就两个:park()和unark(),其中park()方法用来阻塞当前调用线程,unpark()方法用于唤醒指定线程
LockSupport类使用了一种名为Permit(许可)的概念来做到阻塞和唤醒线程的功能,可以把许可看成是一种(0,1)信号量(Semaphore),但与 Semaphore 不同的是,许可的累加上限是1。
初始时,permit为0,当调用unpark()方法时,线程的permit加1,当调用park()方法时,如果permit为0,则调用线程进入阻塞状态。

所以以下代码不会阻塞

// 初始信号量为0,调用unpark,信号量+1
LockSupport.unpark();
// 当前信号量为1,调用park,信号量-1
LockSupport.park();
// 以下代码可以继续执行
doSomething()

2、AQS

AbstractQueueSynchronizer是并发工具的核心,是一个抽象类,提供公平 / 非公平获取锁,获取可重入 / 不可重入锁,共享 / 排他等功能支持
AQS框架,分离了构建同步器时的一系列关注点,它的所有操作都围绕着资源——同步状态(synchronization state)来展开,并替用户解决了如下问题:

  1. 资源是可以被同时访问?还是在同一时间只能被一个线程访问?(共享/独占功能)
  2. 访问资源的线程如何进行并发管理?(等待队列)
  3. 如果线程等不及资源了,如何从等待队列退出?(超时/中断)

这其实是一种典型的模板方法设计模式:父类(AQS框架)定义好骨架和内部操作细节,具体规则由子类去实现。

什么是资源:

同步器 资源的定义
ReentrantLock 资源表示独占锁。State为0表示锁可用;为1表示被占用;为N表示重入的次数
CountDownLatch 资源表示倒数计数器。State为0表示计数器归零,所有线程都可以访问资源;为N表示计数器未归零,所有线程都需要阻塞。
Semaphore 资源表示信号量或者令牌。State≤0表示没有令牌可用,所有线程都需要阻塞;大于0表示由令牌可用,线程每获取一个令牌,State减1,线程没释放一个令牌,State加1。
ReentrantReadWriteLock 资源表示共享的读锁和独占的写锁。state逻辑上被分成两个16位的unsigned short,分别记录读锁被多少线程使用和写锁被重入的次数。

AQS-API
共享和排他

钩子方法 描述
tryAcquire 排它获取(资源数)
tryRelease 排它释放(资源数)
tryAcquireShared 共享获取(资源数)
tryReleaseShared 共享获取(资源数)
isHeldExclusively 是否排它状态
  1. 支持中断超时
  2. 支持独占和共享
  3. 支持Condition条件等待

CAS操作方法
Java中CAS操作的实现都委托给一个名为UnSafe类

方法名 修饰符 描述
compareAndSetState protected final CAS修改同步状态值
compareAndSetHead private final CAS修改等待队列的头指针
compareAndSetTail private final CAS修改等待队列的尾指针
compareAndSetWaitStatus private static final CAS修改结点的等待状态
compareAndSetNext private static final CAS修改结点的next指针

等待队列

方法名 修饰符 描述
enq private 入队操作
addWaiter private 入队操作
setHead private 设置头结点
unparkSuccessor private 唤醒后继结点
doReleaseShared private 释放共享结点
setHeadAndPropagate private 设置头结点并传播唤醒

资源获取操作

方法名 修饰符 描述
cancelAcquire private 取消获取资源
shouldParkAfterFailedAcquire private static 判断是否阻塞当前调用线程
acquireQueued final 尝试获取资源,获取失败尝试阻塞线程
doAcquireInterruptibly private 独占地获取资源(响应中断)
doAcquireNanos private 独占地获取资源(限时等待)
doAcquireShared private 共享地获取资源
doAcquireSharedInterruptibly private 共享地获取资源(响应中断)
doAcquireSharedNanos private 共享地获取资源(限时等待)
acquire public final 独占地获取资源
acquireInterruptibly public final 独占地获取资源(响应中断)
acquireInterruptibly public final 独占地获取资源(限时等待)
acquireShared public final 共享地获取资源
acquireSharedInterruptibly public final 共享地获取资源(响应中断)
tryAcquireSharedNanos public final 共享地获取资源(限时等待)

资源释放:

方法名 修饰符 描述
release public final 释放独占资源
releaseShared public final 释放共享资源

3、等待队列

CLH队列中的结点是对线程的包装,结点一共有两种类型:独占(EXCLUSIVE)和共享(SHARED)。
每种类型的结点都有一些状态,其中独占结点使用其中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2),共享结点使用其中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。

结点状态 描述
CANCELLED 1 取消。表示后驱结点被中断或超时,需要移出队列
SIGNAL -1 发信号。表示后驱结点被阻塞了(当前结点在入队后、阻塞前,应确保将其prev结点类型改为SIGNAL,以便prev结点取消或释放时将当前结点唤醒。)
CONDITION -2 Condition专用。表示当前结点在Condition队列中,因为等待某个条件而被阻塞了
PROPAGATE -3 传播。适用于共享模式(比如连续的读操作结点可以依次进入临界区,设为PROPAGATE有助于实现这种迭代操作。)
INITIAL 0 默认。新结点会处于这种状态
static final class Node {

    // 共享模式结点
    private static final Node SHARED = new Node();

    // 独占模式结点
    private?static final Node EXCLUSIVE = null;

    private?static final int CANCELLED =  1;

    private?static final int SIGNAL    = -1;

    static final int CONDITION = -2;

    static final int PROPAGATE = -3;
    // 等待状态
    volatile int waitStatus;

    // 前驱指针
    volatile Node prev;

    // 后驱指针
    volatile Node next;

    // 结点所包装的线程
    volatile Thread thread;

    // Condition队列使用,存储condition队列中的后继节点
    Node nextWaiter;

    Node() {
    }

    Node(Thread thread, Node mode) {
        this.nextWaiter = mode;
        this.thread = thread;
    }
}

4、加锁

这里以ReentrantLock为例,看看 FairSync 和 NonfairSync 的源码

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

    // 加锁
    final void lock() {
        acquire(1);
    }

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

public final boolean hasQueuedPredecessors() {
    // The correctness of this depends on head being initialized
    // before tail and on head.next being accurate if the current
    // thread is first in queue.
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    // 如果头和尾指向了同一个对象(null)或者头节点下一个为当前节点时,说明队列没有节点,或仅有一个当前节点
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}
  • NonfairSync
/**
 * Sync object for non-fair locks
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

可以发现他们两个都继承自 Sync,Sync继承自AbstractQueuedSynchronizer,在ReentrantLock中的实现如下

  • Sync
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = -5179523762034025860L;

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

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

ReentrantLock的加锁直接委托了Sync的 lock,在Sync中,lock是个抽象方法,依次查看NonfairSync 和 FairSync 的实现,如上文源码注释

非公平锁:

1sync 修改 state (compareAndSetState),尝试直接获取锁,获取成功,则设置排他属性。获取失败,则执行AQS获取锁逻辑

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

2、执行获取锁,tryAcquire (NonfairSync 和 FairSync都有各自的实现),获取失败,获取等待队列,将线程放入等待队列中

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        // 将当前线程包装为独占节点加入队列
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

3、NonfairSync 对 tryAcquire 的实现

  • 获取state状态
  • 如果当前线程持有锁,设置nextc, 添加acquires,并设置给state,可见Reentrant支持可重入锁
  • 如果当前线程没有持有锁,CAS尝试获取锁,获取成功,设置排他性,否则获取锁失败,返回false
// 委托给父类
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}
// 父类实现
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    // 获取state状态
    int c = getState();
    // 如果当前线程持有锁,设置nextc, 添加acquires,并设置给state,可见Reentrant支持可重入锁
    // 如果当前线程没有持有锁,CAS尝试获取锁,获取成功,设置排他性,否则获取锁失败,返回false
    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;
}

3、获取锁失败的处理:acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

  • 为当前线程创建排队节点(这是一个双向链表)
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;
        }
    }
    // 如果尾为null, 先初始化队列
    enq(node);
    return node;
}

// node构造函数
Node(Thread thread, Node mode) {     // Used by addWaiter
    this.nextWaiter = mode;
    this.thread = thread;
}

// 自旋将节点加入尾部,包含队列初始化
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        // Must initialize
        if (t == null) {
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

为已经在队列中的线程以独占不间断模式获取。 由条件等待方法使用以及获取

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 如果当前节点的前驱节点是头节点,当前线程再次获取锁,如果成功,进入if
            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);
    }
}

// 判断是否能阻塞当前线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    // 判断前驱节点状态
    if (ws == Node.SIGNAL)
        // SIGNAL:前驱节点释放锁时,会唤醒当前节点,可以阻塞
        return true;
    if (ws > 0) {
        // CANCELED:前驱节点已中断/取消,需要从队列中移除
        // 循环检查,剔除队列前面无效的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        // 再将当前节点放入队列
        pred.next = node;
    } else {
        // 将前驱节点修改为 SIGNAL, 自旋再次执行此方法时,将走第一条分支
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

// 阻塞当前线程
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

acquireQueued方法抛出异常时会执行 cancelAcquire

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    node.thread = null;

    // Skip cancelled predecessors
    // 再次剔除无效的前驱节点
    Node pred = node.prev;
    while (pred.waitStatus > 0)
         node.prev = pred = pred.prev;

    // 前驱节点的原后继节点,用于后续CAS操作
    Node predNext = pred.next;

    // 当前节点设置为打断
    node.waitStatus = Node.CANCELLED;

    // 如果当前节点为尾节点,设置最后一个有效节点为尾节点
    if (node == tail && compareAndSetTail(node, pred)) {
        // 有效的前驱节点属性next设置为null
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            // 唤醒后继节点
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

5、解锁

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

public final boolean release(int arg) {
    // 尝试解锁
    if (tryRelease(arg)) {
        Node h = head;
        // 释放锁成功唤醒后继节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

释放锁,本质为维护state变量,此处支持可重入锁,如果state值为0,说明释放,取消线程的独占,并更新state

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) {

    // 状态置为0,初始化
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    // 如果 s 为 null, 从后向前迭代,找到最前的未被CANCALLED的节点
    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);
}

后继节点被唤醒后,回到之前的逻辑,开始争夺锁,并将头节点设置为当前节点

6、其他特性

公平锁和非公平锁:

公平锁尝试加锁时,先判断队列中是否有等待线程,如果有,直接进队列

非公平锁直接获取锁,获取失败才进队列

中断特性:

如下代码中,使用了一个bool标记返回标识线程的中断状态,而中断锁会直接抛出异常

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())
                // 中断问题,抛出异常;非中断方法中,返回中断的 bool 变量
                throw new InterruptedException();
        }
    } finally {
        if (failed)
          cancelAcquire(node);
    }
}

限时等待:

public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout);
}

具体实现:

private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    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 true;
            }
            // 自旋时更新剩余等待时间
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                // 超时直接返回获取失败
                return false;
            if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold)
                // 阻塞指定时长,超时则线程自动被唤醒,自旋时,将在上一个if块退出
                // 底层通过 unsafe 实现
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())// 当前线程中断状态
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

7、条件队列

当线程在指定Condition对象上等待的时候,是将线程包装成结点,加入了条件队列,然后阻塞。当线程被通知唤醒时,则是将条件队列中的结点转换成等待队列中的结点,之后的处理就和独占功能完全一样。

J.U.C包提供了Conditon接口,用以对原生的Object.wait()Object.notify()进行增强。

public Condition newCondition() {
        return sync.newCondition();
}
final ConditionObject newCondition() {
    return new ConditionObject();
}

在ReentrantLock中,通过内部ConditionObject实现了Condition接口,提供对条件队列的支持

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    /**
         * First node of condition queue.
     */
    private transient Node firstWaiter;
    /**
     * Last node of condition queue.
     */
    private transient Node lastWaiter;
    /**
     * Creates a new {@code ConditionObject} instance.
     */
    public ConditionObject() {
    }
    ...
}

条件队列操作:

1、加入条件队列等待,条件队列入口

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);
        // 这里被唤醒可能是正常的signal操作也可能是中断
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    /**
     * 走到这里说明节点已经条件满足被加入到了同步队列中或者中断了
     * 和独占锁调用同样的获取锁方法,从这里可以看出条件队列只能用于独占锁
     * 在处理中断之前首先要做的是从同步队列中成功获取锁资源
     */
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    // 走到这里说明已经成功获取到了独占锁,接下来就做些收尾工作
    // 删除条件队列中被取消的节点
    // clean up if cancelled
    if (node.nextWaiter != null)
        unlinkCancelledWaiters();
    // 根据不同模式处理中断
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

addConditionWaiter:将当前线程包装为节点加入条件队列

/**
 * 1.与同步队列不同,条件队列头尾指针是firstWaiter跟lastWaiter
 * 2.条件队列是在获取锁之后,也就是临界区进行操作,因此很多地方不用考虑并发
 */
private Node addConditionWaiter() {
    Node t = lastWaiter;
    //如果最后一个节点被取消,则删除队列中被取消的节点
    //至于为啥是最后一个节点后面会分析
    if (t != null && t.waitStatus != Node.CONDITION) {
        //删除所有被取消的节点
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    //创建一个类型为CONDITION的节点并加入队列,由于在临界区,所以这里不用并发控制
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

删除条件队列当中被取消的节点

/**
 * 删除条件队列当中被取消的节点
 */
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            // 判断中间变量
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        } else
            // 保存的是最靠前的有效条件节点
            trail = t;
        t = next;
    }
}

释放所有资源

/**
 * 入参就是新创建的节点,即当前节点
 */
final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        //这里这个取值要注意,获取当前的state并释放,这从另一个角度说明必须是独占锁
        int savedState = getState();
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            //如果这里释放失败,则抛出异常
            throw new IllegalMonitorStateException();
        }
    } finally {
        /**
         * 如果释放锁失败,则把节点取消,由这里就能看出来上面添加节点的逻辑中
         * 只需要判断最后一个节点是否被取消就可以了
         */
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

判断节点是否在同步队列中

/**
 * 判断节点是否在同步队列中
 */
final boolean isOnSyncQueue(Node node) {
    //快速判断1:节点状态或者节点没有前置节点
    //注:同步队列是有头节点的,而条件队列没有
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    //快速判断2:next字段只有同步队列才会使用,条件队列中使用的是nextWaiter字段
    if (node.next != null) // If has successor, it must be on queue
        return true;
    //上面如果无法判断则进入复杂判断
    return findNodeFromTail(node);
}

2、唤醒等待队列

/**
 * 通知条件队列当中节点到同步队列当中去排队
 */
public final void signal() {
    // 节点不能已经持有独占锁
    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);
}

transferForSignal方法会将CONDITON结点转换为初始结点,并插入【等待队列】

final boolean transferForSignal(Node node) {
    // 尝试转化为初始节点
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

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

8、共享锁

AQS的共享功能,通过钩子方法tryAcquireShared暴露,与独占功能最主要的区别就是:

共享功能的结点,一旦被唤醒,会向队列后部传播(Propagate)状态,以实现共享结点的连续唤醒。这也是共享的含义,当锁被释放时,所有持有该锁的共享线程都会被唤醒,并从等待队列移除。

以CountDownLatch为例:CountDownLatch内部继承了AQS,覆盖了共享获取和释放锁的方法

构造:

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

    Sync(int count) {
        setState(count);
    }

    int getCount() {
        return getState();
    }

    protected int tryAcquireShared(int acquires)
        // 只要 state == 0 就获取锁成功
        return (getState() == 0) ? 1 : -1;
    }

    protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        for (;;) {
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}

private final Sync sync;

await():

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    // 响应中断
    if (Thread.interrupted())
        throw new InterruptedException();
    // 尝试获取共享锁,取决于state变量是否为0(1:获取成功;-1:获取失败)
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

doAcquireSharedInterruptibly

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

setHeadAndPropagate:

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    setHead(node);
    // 判断头节点等待状态
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

doReleaseShared:

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // 如果头节点等待状态为 SIGN, 设置头节点状态归0
                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
        }
        // 退出循环的条件是head并未改变
        if (h == head)                   // loop if head changed
            break;
    }
}

countDown():

public void countDown() {
    // 释放资源
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 释放资源成功后执行
        doReleaseShared();
        return true;
    }
    return false;
}

原文地址:https://www.cnblogs.com/zuier/p/11388795.html

时间: 2024-10-09 19:54:22

图灵学院Java架构师-VIP-并发编程(AQS详解)的相关文章

图灵学院java架构师vip

caier-20 我们一起努力,一起学习,加油! caier-20 微云学习地址:https://share.weiyun.com/5mokPqU 网盘学习地址:https://pan.baidu.com/s/1CTx5SqUeM-ZKtDYLeovODQ 提取码:iclq 致学者:不论你在什么时候开始,重要的是开始之后请不要停止. ava Memory Model,简称JMM,把它抽象成一种规范,用工作内存与主内存这两个概念.在JMM中主内存属于共享数据区域,从某个程度上讲应该包括了堆和方法区

Java架构师VIP课程图灵学院分享

1.熟悉各种框架并且知道他们的原理 2.熟悉虚拟机JVM 并且知道调优方式 能写出更优质的代码 3.连接池等池技术,反射机制 4.nio技术等 5.数据算法结构 sql等 6.熟练linux操作系统 7.各种网络协议 8.集群 9.中间件 图灵学院java架构师课程链接: https://pan.baidu.com/s/1cn5lB7eMSUCsr5CEOkeZTw 提取码 :818j 原文地址:https://www.cnblogs.com/qq-871211680/p/11916918.ht

Java架构师VIP课程一期共89G视频教程 luban it教程

集数合计:14讲Java视频教程详情描述: A0505<Java架构师VIP课程一期共89G视频教程>Java架构师VIP课程一期共89G视频教程 鲁班学院 课件资料全Java视频教程目录: 百度网盘下载点击 ├─撸搬架构├─01.并发专题(一)│      2018-09-11(1)-Java内存模型│      2018-09-13(2)-synchronized原理│      2018-09-16(3)-volatile实现机制│      2018-09-18(4)-DCL-单例模式

鲁班 Java架构师VIP课程一期共89G视频教程 luban it教程

Java视频教程详情描述: A0505<Java架构师VIP课程一期共89G视频教程>Java架构师VIP课程一期共89G视频教程 鲁班学院 课件资料全Java视频教程目录 ├─撸搬架构├─01.并发专题(一)│      2018-09-11(1)-Java内存模型│      2018-09-13(2)-synchronized原理│      2018-09-16(3)-volatile实现机制│      2018-09-18(4)-DCL-单例模式│      2018-09-21(

鲁班学院最新架构师资料《Java架构师VIP课程》

鲁班学院最新架构师资料<Java架构师VIP2019课程一期> 课程目录: ├─01.并发专题(一) │ 2018-09-11(1)-Java内存模型 │ 2018-09-13(2)-synchronized原理 │ 2018-09-16(3)-volatile实现机制 │ 2018-09-18(4)-DCL-单例模式 │ 2018-09-21(5)-并发基础-AQS.CAS2 │ 2018-09-25(6)-并发集合 │ 2018-09-27(7)-原子操作-基本类型 │ 2018-09-3

图灵学院java架构之路-VIP(五)nginx的安装和基本配置

Linux是一套免费使用和自由传播的类Unix操作系统,是一个基于POSIX和Unix的多用户.多任务.支持多线程和多CPU的操作系统.它能运行主要的Unix工具软件.应用程序和网络协议.它支持32位和64位硬件.Linux继承了Unix以网络为核心的设计思想,是一个性能稳定的多用户网络操作系统. 一,安装环境准备: 1.linux 内核2.6及以上版本:使用uname -a查看即可. 2.GCC编辑器. GCC可以用来编译C语言程序.Nginx没有直接提供二级制可执行程序,只能下载源码进行编译

Java并发编程--Volatile详解

摘要 Volatile是Java提供的一种弱同步机制,当一个变量被声明成volatile类型后编译器不会将该变量的操作与其他内存操作进行重排序.在某些场景下使用volatile代替锁可以减少代码量和使代码更易阅读.   Volatile特性 1.可见性:当一条线程对volatile变量进行了修改操作时,其他线程能立即知道修改的值,即当读取一个volatile变量时总是返回最近一次写入的值 2.原子性:对于单个voatile变量其具有原子性(能保证long double类型的变量具有原子性),但对

并发编程Semaphore详解

Semaphore的作用:限制线程并发的数量 位于 java.util.concurrent 下, 构造方法 // 构造函数 代表同一时间,最多允许permits执行acquire() 和release() 之间的代码. public Semaphore(int permits) { sync = new NonfairSync(permits); } //False:表示非公平信号量,即线程启动的顺序与调用semaphore.acquire() 的顺序无关,也就是线程先启动了并不代表先获得 许

39套精品Java从入门到架构师|高并发|高性能|高可用|分布式|集群|电商缓存|性能调优|设计项目实战|视频教程

精品Java高级课,架构课,java8新特性,P2P金融项目,程序设计,功能设计,数据库设计,第三方支付,web安全,高并发,高性能,高可用,分布式,集群,电商,缓存,性能调优,设计模式,项目实战,大型分布式电商项目实战视频教程   视频课程包含: 39套Java精品高级课架构课包含:java8新特性,P2P金融项目,程序设计,功能设计,数据库设计,架构设计,web安全,高并发,高性能,高可用,高可扩展,分布式,集群,电商,缓存,性能调优,设计模式,项目实战,工作流,程序调优,负载均衡,Solr