查看原文
其他

【死磕Java并发】—– J.U.C之AQS:CLH同步队列

2017-12-20 大明哥 芋道源码

此篇博客所有源码均来自JDK 1.8

作者:大明哥 原文地址:http://cmsblogs.com/?p=2188

友情提示:欢迎关注公众号【芋道源码】。😈关注后,拉你进【源码圈】微信群和【大明哥】搞基嗨皮。

友情提示:欢迎关注公众号【芋道源码】。😈关注后,拉你进【源码圈】微信群和【大明哥】搞基嗨皮。

友情提示:欢迎关注公众号【芋道源码】。😈关注后,拉你进【源码圈】微信群和【大明哥】搞基嗨皮。

在上篇博客【死磕Java并发】-----J.U.C之AQS:AQS简介中提到了AQS内部维护着一个FIFO队列,该队列就是CLH同步队列。

CLH同步队列是一个FIFO双向队列,AQS依赖它来完成同步状态的管理,当前线程如果获取同步状态失败时,AQS则会将当前线程已经等待状态等信息构造成一个节点(Node)并将其加入到CLH同步队列,同时会阻塞当前线程,当同步状态释放时,会把首节点唤醒(公平锁),使其再次尝试获取同步状态。

在CLH同步队列中,一个节点表示一个线程,它保存着线程的引用(thread)、状态(waitStatus)、前驱节点(prev)、后继节点(next),其定义如下:

  1. static final class Node {

  2.    /** 共享 */

  3.    static final Node SHARED = new Node();

  4.    /** 独占 */

  5.    static final Node EXCLUSIVE = null;

  6.    /**

  7.     * 因为超时或者中断,节点会被设置为取消状态,被取消的节点时不会参与到竞争中的,他会一直保持取消状态不会转变为其他状态;

  8.     */

  9.    static final int CANCELLED =  1;

  10.    /**

  11.     * 后继节点的线程处于等待状态,而当前节点的线程如果释放了同步状态或者被取消,将会通知后继节点,使后继节点的线程得以运行

  12.     */

  13.    static final int SIGNAL    = -1;

  14.    /**

  15.     * 节点在等待队列中,节点线程等待在Condition上,当其他线程对Condition调用了signal()后,改节点将会从等待队列中转移到同步队列中,加入到同步状态的获取中

  16.     */

  17.    static final int CONDITION = -2;

  18.    /**

  19.     * 表示下一次共享式同步状态获取将会无条件地传播下去

  20.     */

  21.    static final int PROPAGATE = -3;

  22.    /** 等待状态 */

  23.    volatile int waitStatus;

  24.    /** 前驱节点 */

  25.    volatile Node prev;

  26.    /** 后继节点 */

  27.    volatile Node next;

  28.    /** 获取同步状态的线程 */

  29.    volatile Thread thread;

  30.    Node nextWaiter;

  31.    final boolean isShared() {

  32.        return nextWaiter == SHARED;

  33.    }

  34.    final Node predecessor() throws NullPointerException {

  35.        Node p = prev;

  36.        if (p == null)

  37.            throw new NullPointerException();

  38.        else

  39.            return p;

  40.    }

  41.    Node() {

  42.    }

  43.    Node(Thread thread, Node mode) {

  44.        this.nextWaiter = mode;

  45.        this.thread = thread;

  46.    }

  47.    Node(Thread thread, int waitStatus) {

  48.        this.waitStatus = waitStatus;

  49.        this.thread = thread;

  50.    }

  51. }

CLH同步队列结构图如下:

入列

学了数据结构的我们,CLH队列入列是再简单不过了,无非就是tail指向新节点、新节点的prev指向当前最后的节点,当前最后一个节点的next指向当前节点。代码我们可以看看addWaiter(Node node)方法:

  1.    private Node addWaiter(Node mode) {

  2.        //新建Node

  3.        Node node = new Node(Thread.currentThread(), mode);

  4.        //快速尝试添加尾节点

  5.        Node pred = tail;

  6.        if (pred != null) {

  7.            node.prev = pred;

  8.            //CAS设置尾节点

  9.            if (compareAndSetTail(pred, node)) {

  10.                pred.next = node;

  11.                return node;

  12.            }

  13.        }

  14.        //多次尝试

  15.        enq(node);

  16.        return node;

  17.    }

addWaiter(Node node)先通过快速尝试设置尾节点,如果失败,则调用enq(Node node)方法设置尾节点

  1.    private Node enq(final Node node) {

  2.        //多次尝试,直到成功为止

  3.        for (;;) {

  4.            Node t = tail;

  5.            //tail不存在,设置为首节点

  6.            if (t == null) {

  7.                if (compareAndSetHead(new Node()))

  8.                    tail = head;

  9.            } else {

  10.                //设置为尾节点

  11.                node.prev = t;

  12.                if (compareAndSetTail(t, node)) {

  13.                    t.next = node;

  14.                    return t;

  15.                }

  16.            }

  17.        }

  18.    }

在上面代码中,两个方法都是通过一个CAS方法compareAndSetTail(Node expect, Node update)来设置尾节点,该方法可以确保节点是线程安全添加的。在enq(Node node)方法中,AQS通过“死循环”的方式来保证节点可以正确添加,只有成功添加后,当前线程才会从该方法返回,否则会一直执行下去。

过程图如下:

出列

CLH同步队列遵循FIFO,首节点的线程释放同步状态后,将会唤醒它的后继节点(next),而后继节点将会在获取同步状态成功时将自己设置为首节点,这个过程非常简单,head执行该节点并断开原首节点的next和当前节点的prev即可,注意在这个过程是不需要使用CAS来保证的,因为只有一个线程能够成功获取到同步状态。过程图如下:

参考资料

Doug Lea:《Java并发编程实战》 方腾飞:《Java并发编程的艺术》




    您可能也对以下帖子感兴趣

    文章有问题?点此查看未经处理的缓存