枯燥的J.U.C - AbstractQueuedSynchronizer(二)

俗话说得好,编程不识Doug Lea,写尽Java也枉然。我们在上个章节初步分析了AQS(独占模式)同步等待队列的实现过程,这章我们将继续以J.U.C的基础API为切入点,分别对AQS的条件等待队列与共享锁实现进行源码分析。

image.png

~~~~~~~~~~~~~ 祖师爷庇护,并发从此无难题。~~~~~~~~~~~~

AQS中的ConditionObject

还记得我们之前在Node类中看到的 waitStatus属性,它有个CONDITION状态表示现在该节点是在Condition等待队列中,根据 waitStatus的不同,我们可以把它看成两条队列,等待获取锁资源的同步等待队列(waitStatus=0),等待唤醒的Condition队列。


image.png

等待队列在J.U.C下的BlockingQueue中的使用(本章主要探讨条件队列的实现原理及应用场景,部分与本章无关的代码就不贴出):

    // 队列中的当前元素数
    private final AtomicInteger count = new AtomicInteger();

    // 出队锁
    private final ReentrantLock takeLock = new ReentrantLock();
    // 出队等待条件
    private final Condition notEmpty = takeLock.newCondition();

    // 入队锁
    private final ReentrantLock putLock = new ReentrantLock();
    // 入队等待条件
    private final Condition notFull = putLock.newCondition();

基于生产-消费模式,put生产,take消费,根据维护的当前count值来判断是否需要进入条件等待或被唤醒转移到CLH队列去尝试获取锁,大致逻辑如下:

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // 本地标量
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            // 容量满了 加入条件等待队列
            while (count.get() == capacity) {
                notFull.await();
            }
            // 被唤醒并满足条件后加入元素操作
            enqueue(node);
            // getAndIncrement() 操作会返回当前修改对象的旧值
            c = count.getAndIncrement();
            // 容量未慢,继续唤醒因条件等待而阻塞的put线程
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        // 若之前容量为0,说明已经有take线程进入条件等待队列,去唤醒它
        if (c == 0)
            signalNotEmpty();
    }

值得注意的是:LinkedBlockingQueue之所以能够高效的处理并发数据,是因为它的生产者端和消费者端分别采用了独立的锁来控制数据同步,这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据,以此来提高整个队列的并发性能。

根据源码上的注释,我们大概明白了它的 put()流程,而 take()与之逻辑相反,这里就不再赘述,我们本章重点是 await() 与 signal()方法。

他们是AbstractQueuedSynchronizer.ConditionObject中用来控制线程条件等待的方法:await()阻塞线程进入条件等待队列,signal()唤醒条件等待队列中阻塞的线程。功能类似与 synchronized块中的 wait()、notify()与notifyAll()方法(我们之前讲的Lock也是当时李大爷对JDK1.5之前 synchronized 功能的Java版改造,使我们用起来更灵活,也更加轻量级)。

下面我们直接看 await()方法,释放占有的锁资源,使自己加入条件队列并阻塞,且被唤醒后继续竞争锁资源的逻辑:

        public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            // 当前节点加入条件队列
            Node node = addConditionWaiter();
            // 完全释放锁资源
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            // 首次自旋当前线程已经是非CLH节点,被阻塞
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                // 这里被唤醒可能是正常的signal操作也可能是中断
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            // 被唤醒后尝试在CLH队列中获取锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

其中的 addConditionWaiter()方法,我们在之前已经提到过,它是由Node节点中 nextWaiter属性来构建起的一个单向链表结构,这里不再赘述。

我们接着看 signal()方法,通知唤醒条件队列中所有节点的逻辑:

        public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            // 发信号通知条件队列的节点准备到CLH队列当中去排队
            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);
        }

AbstractQueuedSynchronizer中的 transferForSignal()方法把条件等待节点转移到CLH队列:

    final boolean transferForSignal(Node node) {
        // 由条件等待标识修改为等待竞争锁标识
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;

        // 转移到CLH队列中并返回它的前驱节点
        Node p = enq(node);
        int ws = p.waitStatus;
        // 前驱节点不可用 或者 修改信号量状态失败
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }
AQS条件队列的执行过程分析:
  1. 在操作条件队列之前首先需要成功获取独占锁,不然直接在获取独占锁的时候就已经被挂起。
  2. 成功获取独占锁以后,如果当前条件还不满足,则在当前锁的条件队列上挂起,与此同时释放掉当前获取的锁资源(若不释放会造成资源锁死)。
  3. 如果被唤醒,则检查是否可以获取独占锁,否则继续挂起。

条件满足后的唤醒过程:
把当前等待队列中的第一个有效节点(如果被取消就无效了)加入同步队列等待被前置节点唤醒,如果此时前置节点被取消,则直接唤醒该节点让它重新在同步队列里适当的尝试获取锁或者挂起。

AQS共享锁之Semaphore

Semaphore 是 Lock 的加强版,它的作用是控制访问特定资源的线程数目,底层依赖AQS的状态 state,是在生产当中比较常用的限流工具类,应用场景如下例图:


image.png

如何使用Semaphore?

        // 入参表示同一时刻只允许 3个线程进入,其他线程只能等待。
        Semaphore semaphore = new Semaphore(3);
        new Thread(()->{
            try {
                semaphore.acquire();
                // 此处业务逻辑。。。
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

注意它的构造,默认也是非公平锁,最终会把 permits入参赋值给AQS的 state属性,通过对 state资源的同步加减来实现并发限流控制

关键方法:

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

通过上面代码中的 addWaiter(Node.SHARED)看出,它也是使用的CLH队列,Node构造为共享模式。因为我们已经搞清楚了AQS的CLH队列与条件队列的核心逻辑,对于这些基于它的API实现,我们理解起来也更加容易,这里就不再一步一步贴代码。

AQS共享锁之CountDownLatch

CountDownLatch是通过一个计数器来实现的,计数器的初始值为线程的数量。每当一个线程完成了自己的任务后,计数器的值就会减1。当计数器值到达0时,它表示所有的线程已经完成了任务,然后在闭锁上等待的线程就可以恢复执行任务。

如何使用CountDownLatch?

        CountDownLatch countDownLatch = new CountDownLatch(2); // 两个执行子线程(任务)
        new Thread(()->{
            countDownLatch.countDown(); // 线程一执行完毕 state-1
        }).start();
        new Thread(()->{
            countDownLatch.countDown(); // 线程二执行完毕 state-1
        }).start();
        // 主线程等待分支线程执行完毕(并返回结果)
        countDownLatch.await(); // 当前state=0
        System.out.println("分支汇总执行完毕");

AQS共享锁之CyclicBarrier

栅栏屏障,让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续运行。

        // 入参代表屏障拦截数量
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
        new Thread(()->{
            try {
                cyclicBarrier.await();
                System.out.println("分支一计算执行完毕");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(()->{
            try {
                Thread.sleep(2000);
                cyclicBarrier.await();
                System.out.println("分支二计算执行完毕");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }).start();
        cyclicBarrier.await();
        // 注意:主线程可能先于分支线程的后续任务逻辑
        System.out.println("主分支汇总完毕"); 
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 211,123评论 6 490
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,031评论 2 384
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 156,723评论 0 345
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,357评论 1 283
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,412评论 5 384
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,760评论 1 289
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,904评论 3 405
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,672评论 0 266
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,118评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,456评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,599评论 1 340
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,264评论 4 328
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,857评论 3 312
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,731评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,956评论 1 264
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,286评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,465评论 2 348