juc之AQS源码二

发表于 更新于 分类于 ， ， ，
本文字数： 3k 阅读时长 ≈ 3 分钟

共享方式

  juc之AQS源码一中讲了独占方式的源码，这里分析共享方式的源码。

acquireShared共享方式获取资源

acquireShared方法

  acquireShared方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源，获取成功则直接返回，获取失败则进入等待队列，直到获取到资源为止，整个过程忽略中断。

1
2
3
4
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

  这里tryAcquireShared()依然需要自定义同步器去实现。但是AQS已经把其返回值的语义定义好了：负值代表获取失败；0代表获取成功，但没有剩余资源；正数表示获取成功，还有剩余资源，其他线程还可以去获取。
  所以函数流程如下：

  1. tryAcquireShared()尝试获取资源，成功则直接返回。
  2. 失败则通过doAcquireShared()进入等待队列，直到获取到资源为止才返回。

doAcquireShared方法

  此方法用于将当前线程加入等待队列尾部休息，直到其他线程释放资源唤醒自己，自己成功拿到相应量的资源后才返回。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
private void doAcquireShared(int arg) {
    // 加入队列尾部
    final Node node = addWaiter(Node.SHARED);
    // 是否成功的标志
    boolean failed = true;
    try {
        // 是否有外部中断的标志
        boolean interrupted = false;
        for (;;) {
            // 拿到前驱节点
            final Node p = node.predecessor();
            // 如果到head的下一个，因为head是拿到资源的线程，此时node被唤醒，很可能是head用完资源来唤醒自己的
            if (p == head) {
                // 尝试获取资源
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 如果有剩余资源唤醒后面的线程，这里就是和独占式的区别
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    // 如果有外部中断，补充中断
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

  这个方法和acquireQueued方法很相似，只不过这里将补充中断的位置搬进来了。

setHeadAndPropagate方法

  这个方法在自己苏醒的同时，如果条件符合（比如还有剩余资源），还会去唤醒后继结点，毕竟是共享模式！

1
2
3
4
5
6
7
8
9
10
11
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    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();
    }
}

总结

  基本流程：

  1. tryAcquireShared()尝试获取资源，成功则直接返回。
  2. 失败则通过doAcquireShared()进入等待队列park()，直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

  跟acquire的流程相似，多了个自己拿到资源后，还回去唤醒后面的线程。

releaseShared共享方式释放资源

releaseShared方法

  此方法是共享模式下线程释放共享资源的顶层入口。
  它会释放指定量的资源，如果成功释放且允许唤醒等待线程，它会唤醒等待队列里的其他线程来获取资源。

1
2
3
4
5
6
7
8
9
public final boolean releaseShared(int arg) {
    // 尝试释放资源
    if (tryReleaseShared(arg)) {
        // 唤醒后面的线程
        doReleaseShared();
        return true;
    }
    return false;
}

  释放掉资源后，唤醒后继。
  跟独占模式下的release()相似，但有一点稍微需要注意：独占模式下的tryRelease()在完全释放掉资源（state=0）后，才会返回true去唤醒其他线程，这主要是基于独占下可重入的考量；而共享模式下的releaseShared()则没有这种要求（只要有释放，就回去唤醒），共享模式实质就是控制一定量的线程并发执行，那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。

doReleaseShared方法

  这个方法用于唤醒后继节点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // 该状态表示后继节点需要唤醒
            if (ws == Node.SIGNAL) {
                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
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

参考文献 & 鸣谢

相关文章