CountDownLatch、CyclicBarrier和Semaphore 使用示例及原理

也欢迎查看我的csdn的此篇文章链接:CountDownLatch、CyclicBarrier和Semaphore 使用示例及原理

CountDownLatch

CountDownLatch用户监听某些初始化操作,并且线程进行阻塞,等初始化执行完毕后,通知主线程继续工作执行。

CountDownLatch 使用示例

使用示例,线程t3 要等待t1和t2执行完毕才执行:

/**
 * @Description: CountDownLatch 等待和唤醒
 * @Author: wangmeng
 * @Date: 2018/12/16-16:38
 */
public class UseCountDownLatch {

    public static void main(String[] args) {
        CountDownLatch countDownLatch = new CountDownLatch(2);
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("进入t1线程。。。");
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                System.out.println("t1线程初始化完毕,通知t3线程继续操作!");
                countDownLatch.countDown();
            }
        }, "t1");

        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("进入t2线程。。。");
                try {
                    TimeUnit.SECONDS.sleep(4);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                System.out.println("t2线程初始化完毕,通知t3线程继续操作!");
                countDownLatch.countDown();
            }
        }, "t2");

        Thread t3 = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("进入t3 线程,并且等待...");
                try {
                    countDownLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                System.out.println("t3线程进行后续的执行操作...");
            }
        }, "t3");

        t1.start();
        t2.start();
        t3.start();
    }
}

打印结果:

进入t1线程。。。
进入t3 线程,并且等待...
进入t2线程。。。
t1线程初始化完毕,通知t3线程继续操作!
t2线程初始化完毕,通知t3线程继续操作!
t3线程进行后续的执行操作...

CountDownLatch 源码解读

其实CountDownLatch用的底层原理就是AQS, 可以参考:(AQS原理详解)。AQS全局维护的有一个volatile修饰的state字段,当state为0时就会通知countDownLatch等待线程执行。 这也就是所以我们在new CountDownLatch(int n) 时指定的参数,n为多少,也就是要调用多少次countDown()方法。

public void await() throws InterruptedException { }; //调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; //和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行
public void countDown() { }; //将count值减1

看看await()方法, 其底层调用的就是AQS中的getState方法,通过判断state是否为0来决定是否唤醒等待的线程。 如果不为0则调用Unsafe中的park方法进行自旋,直到state==0时才继续往下执行(唤醒等待的线程)。

public void await() throws InterruptedException {
    //调用AQS中的方法
     sync.acquireSharedInterruptibly(1);
 }

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

//CountDownLatch中的方法,获取state值
protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
 }

//当AQS中的state不为0就会执行此方法,这个方法也就是让线程等待。使用直到state==0才结束循环。
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);
        }
    }

看到上面await方法了,那么countDown就可以直接猜出来了,无外乎就是使得AQS中的state通过CAS操作进行减一,如下:

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

CyclicBarrier

CyclicBarrier是回环栅栏的概念,多线程来的进行阻塞,等待某一个临界值条件满足后,同时执行。 假设有一个场景:每个线程代表一个跑步运动员,当运动员都准备好后,才一起出发,只要有一个人没有准备好,大家都等待。

CyclicBarrier 应用实例

/**
 * @Description: 测试CyclicBarrier
 * @Author: wangmeng
 * @Date: 2018/12/16-17:05
 */
public class UseCyclicBarrier {

    //模拟运动员的类。
    static class Runner implements Runnable {

        private String name;

        private CyclicBarrier cyclicBarrier;

        @Override
        public void run() {
            try {
                System.out.println("运动员:" + this.name + "进行准备工作!");
                TimeUnit.SECONDS.sleep((new Random().nextInt(5)));
                System.out.println("运动员:" + this.name + "准备完成!");
                this.cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }

            System.out.println("运动员" + this.name + "开始起跑!!!");
        }

        public Runner(String name, CyclicBarrier cyclicBarrier) {
            this.name = name;
            this.cyclicBarrier = cyclicBarrier;
        }
    }

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
        ExecutorService executorPools = Executors.newFixedThreadPool(3);

        executorPools.submit(new Thread(new Runner("张三", cyclicBarrier)));
        executorPools.submit(new Thread(new Runner("李四", cyclicBarrier)));
        executorPools.submit(new Thread(new Runner("王五", cyclicBarrier)));

        executorPools.shutdown();
    }
}

打印的结果:

运动员:张三进行准备工作!
运动员:李四进行准备工作!
运动员:王五进行准备工作!
运动员:张三准备完成!
运动员:王五准备完成!
运动员:李四准备完成!
运动员李四开始起跑!!!
运动员张三开始起跑!!!
运动员王五开始起跑!!!

可以看到三个线程都是先执行完初始化操作,然后才一起唤醒执行后续的操作。

CyclicBarrier 源码解读

CyclicBarrier是通过ReentrantLock和Condition来实现的。

private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    final ReentrantLock lock = this.lock;
    // 锁住
    lock.lock();
    try {
        // 当前代
        final Generation g = generation;
        // 如果这代损坏了,抛出异常
        if (g.broken)
            throw new BrokenBarrierException();

        // 如果线程中断了,抛出异常
        if (Thread.interrupted()) {
            // 将损坏状态设置为 true
            // 并通知其他阻塞在此栅栏上的线程
            breakBarrier();
            throw new InterruptedException();
        }
        // 获取下标    
        int index = --count;
        // 如果是 0 ,说明到头了
        if (index == 0) { // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                // 执行栅栏任务
                if (command != null)
                    command.run();
                ranAction = true;
                // 更新一代,将 count 重置,将 generation 重置.
                // 唤醒之前等待的线程
                nextGeneration();
                // 结束
                return 0;
            } finally {
                // 如果执行栅栏任务的时候失败了,就将栅栏失效
                if (!ranAction)
                    breakBarrier();
            }
        }

        for (;;) {
            try {
                // 如果没有时间限制,则直接等待,直到被唤醒
                if (!timed)
                    trip.await();
                // 如果有时间限制,则等待指定时间
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                // g == generation >> 当前代
                // ! g.broken >>> 没有损坏
                if (g == generation && ! g.broken) {
                    // 让栅栏失效
                    breakBarrier();
                    throw ie;
                } else {
                    // 上面条件不满足,说明这个线程不是这代的.
                    // 就不会影响当前这代栅栏执行逻辑.所以,就打个标记就好了
                    Thread.currentThread().interrupt();
                }
            }
            // 当有任何一个线程中断了,会调用 breakBarrier 方法.
            // 就会唤醒其他的线程,其他线程醒来后,也要抛出异常
            if (g.broken)
                throw new BrokenBarrierException();
            // g != generation >>> 正常换代了
            // 一切正常,返回当前线程所在栅栏的下标
            // 如果 g == generation,说明还没有换代,那为什么会醒了?
            // 因为一个线程可以使用多个栅栏,当别的栅栏唤醒了这个线程,就会走到这里,所以需要判断是否是当前代。
            // 正是因为这个原因,才需要 generation 来保证正确。
            if (g != generation)
                return index;
            // 如果有时间限制,且时间小于等于0,销毁栅栏,并抛出异常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

用上面的示例总结一下CyclicBarrier的await方法实现,假设线程thread1和线程thread2都执行到CyclicBarrier的await(),都进入dowait(boolean timed, long nanos),thread1先获取到独占锁,执行到--count的时,index等于1,所以进入下面的for循环,接着执行trip.await(),进入await()方法,执行Node node = addConditionWaiter()将当前线程构造成Node节点并加入到Condition等待队列中,然后释放获取到的独占锁,当前线程进入阻塞状态;此时,线程thread2可以获取独占锁,继续执行--count,index等于0,所以先执行command.run(),输出myThread,然后执行nextGeneration(),nextGeneration()中trip.signalAll()只是将Condition等待队列中的Node节点按之前顺序都转移到了AQS同步队列中,这里也就是将thread1对应的Node节点转移到了AQS同步队列中,thread2执行完nextGeneration(),返回return 0之前,细看代码还需要执行lock.unlock(),这里会执行到ReentrantLock的unlock()方法,最终执行到AQS的unparkSuccessor(Node node)方法,从AQS同步队列中的头结点开始释放节点,唤醒节点对应的线程,即thread1恢复执行。

如果有三个线程thread1、thread2和thread3,假设线程执行顺序是thread1、thread2、thread3,那么thread1、thread2对应的Node节点会被加入到Condition等待队列中,当thread3执行的时候,会将thread1、thread2对应的Node节点按thread1、thread2顺序转移到AQS同步队列中,thread3执行lock.unlock()的时候,会先唤醒thread1,thread1恢复继续执行,thread1执行到lock.unlock()的时候会唤醒thread2恢复执行。

更多可参考:并发编程之 CyclicBarrier 源码分析

CountdownLatch和CyclicBarrier的区别

1、CountDownLatch简单的说就是一个线程等待,直到他所等待的其他线程都执行完成并且调用countDown()方法发出通知后,当前线程才可以继续执行。 2、CyclicBarrier是所有线程都进行等待,直到所有线程都准备好进入await()方法之后,所有线程同时开始执行! 3、CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景,比如如果计算发生错误,可以重置计数器,并让线程们重新执行一次。

Semaphore

Semaphore翻译成字面意思为 信号量,Semaphore可以控同时访问的线程个数,通过 acquire() 获取一个许可,如果没有就等待,而 release() 释放一个许可。 关于限流的其他方式可以参考我另一篇博文:限流的简单使用及学习 相关概念:

  • PV(page view)网站的总访问量,页面浏览量或点击量,用户每刷新一次就会被记录一次。
  • UV(unique Visitor)访问网站的一台电脑客户端为一个访客。一般来讲时间上以00:00-24:00之内相同ip的客户端只记录。
  • QPS(query per second)即每秒查询数,qps很大程度上代表了系统业务上的繁忙程度,每次请求的背后,可能对应着多次磁盘I/O,多次网络请求,多个cpu时间片等。我们通过qps可以非常直观的了解当前系统业务情况,一旦当前qps超过所设定的预警阀值,可以考虑增加机器对集群扩容,以免压力过大导致宕机,可以根据前期的压力测试得到估值,在结合后期综合运维情况,估算出阀值。
  • RT(response time)请求的响应时间,这个指标非常关键,直接说明前端用户的体验,任何系统设计师都想降低rt时间。
  • 当然还涉及cpu、内存、网络、磁盘等情况,更细节的问题很多,如select、update、delete/ps等数据库层面的统计。
  • 容量评估:一般来说通过开发、运维、测试、以及业务等相关人员,综合出系统的一系列阀值,然后我们根据关键阀值如qps、rt等,对系统进行有效的变更。
  • 一般来讲,我们进行多轮压力测试以后,可以对系统进行峰值评估,采用所谓的80/20原则,即80%的访问请求将在20%的时间内达到。这样我们可以根据系统对应的PV计算出峰值qps。
  • 峰值qps= (总PV × 80%)/ (60 × 60 × 24 × 20%)
  • 然后在将总的峰值qps除以单台机器所能承受的最高的qps值,就是所需要机器的数量:机器数 = 总的峰值qps / 压测得出的单机极限qps
  • 当然不排除系统在上线前进行大型促销活动,或者双十一、双十二热点事件、遭受到DDos攻击等情况,系统的开发和运维人员急需要了解当前系统运行的状态和负载情况,一般都会有后台系统去维护。

Semaphore 使用示例:

/**
 * @Description:使用Semaphore模拟限流操作
 * @Author: wangmeng
 * @Date: 2018/12/16-18:30
 */
public class UseSemaphore {

    public static void main(String[] args) {
        ExecutorService threadPools = Executors.newFixedThreadPool(20);
        //同一时间只能有5个线程执行
        Semaphore semaphore = new Semaphore(5);


        for (int i = 0; i < 20; i++) {
            final int token = i;
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        semaphore.acquire();

                        //进行业务操作
                        System.out.println("获得许可,执行操作..." + token);
                        long sleepTime = (long)(Math.random() * 10000);
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        semaphore.release();
                    }
                }
            };

            threadPools.execute(run);
        }

        System.out.println("queue length: " + semaphore.getQueueLength());
        threadPools.shutdown();
    }
}

原理也是使用AQS中的state变量。代码我就不贴了。 Semaphore 就是一个共享锁,通过设置 state 变量来实现对这个变量的共享。当调用 acquire 方法的时候,state 变量就减去一,当调用 release 方法的时候,state 变量就加一。当 state 变量为 0 的时候,别的线程就不能进入代码块了,就会在 AQS 中阻塞等待。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券