并发编程系列之CountDownLatch对战Cyclicbarrier

本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer,已开源在github ,欢迎star!

转载声明:转载请注明出处,本技术博客是本人原创文章

本文GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收录,这是我花了6个月总结的一线大厂Java面试总结,本人已拿大厂offer,欢迎star

原文链接:blog.ouyangsihai.cn >> 并发编程系列之CountDownLatch对战Cyclicbarrier

并发编程系列之CountDownLatch对战Cyclicbarrier

前言

前面我们介绍了并发容器和队列,今天我们来介绍几个非常有用的并发工具类,今天主要讲CountDownLatch和Cyclicbarrier这两个工具类,通过讲解并对比两个类的区别,OK,让我们开始今天的并发之旅吧。

什么是CountDownLatch?

CountDownLatch用于监听某些初始化操作,等待初始化执行完毕,通知主线程继续工作,允许一个或者多个线程等待其他线程完成操作。之前我们知道要实现线程等待还有一个方法就是jion方法,先让我们来回忆什么是Join方法:

Join用于让当前执行线程等待Join线程执行结束,实现原理是,不停的检查Join线程是否存活,如果存活则让当前线程永远等待下去,如果Join线程终止,则调用this.notifyAll方法唤醒等待的线程;

CountDownLatch其实也是来做这件事的,而且比Join更强大,使用起来也很轻便。

如何使用CountDownLatch?

我们看下面这个demo,看看如何使用CountDownLatch:


public static void main(String[] args) {
    // CountDownLatch接收一个int类型的计算器,此处是2代表计数器为2,意思是需要等待2个线程唤醒
    final CountDownLatch countDown = new CountDownLatch(2);
    
    Thread t1 = new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          System.out.println("进入线程t1" + "等待其他线程处理完成...");
          // countDown.await()方法会阻塞当前线程即t1,没执行一次countDown()方法计数器就会-1
          // 直到计数器=0,则当前阻塞的线程t1被唤醒,继续执行
          countDown.await();
          System.out.println("t1线程继续执行...");
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    },"t1");
    
    Thread t2 = new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          System.out.println("t2线程进行初始化操作...");
          Thread.sleep(3000);
          System.out.println("t2线程初始化完毕,通知t1线程继续...");
          // 计数器-1
          countDown.countDown();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
    Thread t3 = new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          System.out.println("t3线程进行初始化操作...");
          Thread.sleep(4000);
          System.out.println("t3线程初始化完毕,通知t1线程继续...");
          // 计数器再-1,唤醒t1,t1继续执行
          countDown.countDown();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
    t1.start();
    t2.start();
    t3.start();
  }

执行结果:

并发编程系列之CountDownLatch对战Cyclicbarrier

猜想:假设t1或者t2由于某某原因发生异常未能执行countDown.countDown()那么,t1线程岂不是要一直处于等待状态吗?当然JDK的设计大佬们才不会给你留下这么明显的问题呢,所以countDown还提供了一个


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

,这个方法会在特定时间后,结束阻塞的线程。

CountDownLatch底层分析

我们主要看下CountDownLatch的await方法和countDown方法的源码,首先看看await源码:await内部采用公平锁来实现等待


public void await() throws InterruptedException {
        // 采用公平锁机制
        sync.acquireSharedInterruptibly(1);
}
public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

再看下acquireSharedInterruptibly,这里只分析await,超时await原理也差不多:


public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 判断是否发生中断
        if (Thread.interrupted())
            throw new InterruptedException();
        // -1表示获取到了共享锁,1表示没有获取共享锁
        if (tryAcquireShared(arg)  0)
            // 获取共享锁,继续执行
            doAcquireSharedInterruptibly(arg);
    }

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

再看下countDown方法:


public void countDown() {
        // 每次释放一个计数器
        sync.releaseShared(1);
}

 public final boolean releaseShared(int arg) {
        //尝试释放共享锁  
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }


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;            
               // 循环检查
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                
            // loop on failed CAS
            }
            if (h == head)                   
      // loop if head changed
                break;
        }
    }

什么是Cyclicbarrier?

Cyclicbarrier指的是可循环使用的屏障,主要是让一组线程到达一个屏障之后被阻塞,当最后一个线程到达时,屏障才会开门,所有被屏障拦截的线程才会继续干活。

如何使用Cyclicbarrier?


static class Runner implements Runnable {  
      private CyclicBarrier barrier;  
      private String name;  
      
      public Runner(CyclicBarrier barrier, String name) {  
          this.barrier = barrier;  
          this.name = name;  
      }  
      @Override  
      public void run() {  
          try {  
            // 因为是先打印后阻塞,所以这里getNumberWaiting的+1
            int numberWaiting = barrier.getNumberWaiting();
            int count = numberWaiting + 1 ;
            System.out.println(name + " 进入赛道,签到完毕,当前人数"+count);  
              barrier.await();  
          } catch (InterruptedException e) {  
              e.printStackTrace();  
          } catch (BrokenBarrierException e) {  
              e.printStackTrace();  
          }  
          System.out.println(name + " Go!!");  
      }  
  } 
  
    public static void main(String[] args) throws IOException, InterruptedException {  
        CyclicBarrier barrier = new CyclicBarrier(10);
       // Executors是我们后续会讲的线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);  
        
        for (int i = 100; i  110; i++) {
          Thread.sleep(1000);  
           executor.submit(new Thread(new Runner(barrier, i+"号选手进场")));  
    }
        executor.shutdown();  
    }

执行结果:

并发编程系列之CountDownLatch对战Cyclicbarrier

某些情况下,我们需要让阻塞屏障解除的时候,某些线程需要先执行,例如某个运动员买通了裁判,比赛开始时,比别的选手提前开跑,当然这在现实比赛中是不允许的,此处我只是打个比方,对于这样的场景,Cyclicbarrier提供了:


public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties = 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

用于在线程到达屏障时,优先执行barrierAction线程;

Cyclicbarrier底层实现


public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen;
        }
    }


    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        // 使用重入锁,同步进行wait操作,计数器+1
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Generation g = generation;
            // 当前Generation处于打破状态,抛出异常
            if (g.broken)
                throw new BrokenBarrierException();
            // 当前Generation处于中断状态,抛出异常,并重置计数器,唤醒所有等待线程,可见见下面源码
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

           int index = --count;
           // 当最后一个线程也到达了,就从调用中返回
           if (index == 0) { 
               boolean ranAction = false;
               try {
                   final Runnable command = barrierCommand;
                   if (command != null)
                       command.run();
                   ranAction = true;
                   nextGeneration();
                   return 0;
               } finally {
                   // 如果运行command失败也会导致当前屏障被打破
                   if (!ranAction)
                       breakBarrier();
               }
           }

            // loop until tripped, broken, interrupted, or timed out
            for (;;) {
                try {
                    if (!timed)
                        trip.await();
                    else if (nanos  0L)
                       // 挂起在条件变量的等待队列里,等待信号并自动释放锁
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    // 如果当前线程被中断了则使得屏障被打破。并抛出异常
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        Thread.currentThread().interrupt();
                    }
                }
                //从阻塞恢复之后,需要重新判断当前的状态
                if (g.broken)
                    throw new BrokenBarrierException();

                if (g != generation)
                    return index;

                if (timed && nanos = 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }



 private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }


CountDownLatch和Cyclicbarrier比较

CountDownLatch就像一场跑步比赛,假设这场比赛有10个运动员,那么计数器初始值就为10,裁判员喊下比赛开始,就await阻塞在那,当每个运动员跑到终点就countDown一次,计数器-1,知道最后一个运动员到达终点即计数器为0,此时裁判员被唤醒,统计比赛结果,完成比赛。

Cyclicbarrier就像这场比赛时,裁判员首先准备好10条赛道,准备完毕就拿个小本子在那等着,每当以为选手到达赛道就签到一次,当10个选手全部签到完毕,裁判员就宣布比赛正式开始,继续执行下面的比赛。如果中间因为某某原因,某个选手未能到场或者天气原因,比赛推迟,签到信息就重置,比赛恢复之后选手需要重新签到;

区别总结:CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景,比如如果计算发生错误,可以重置计数器,并让线程们重新执行一次。

原文始发于微信公众号(Justin的后端书架):

本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer,已开源在github ,欢迎star!

转载声明:转载请注明出处,本技术博客是本人原创文章

本文GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收录,这是我花了6个月总结的一线大厂Java面试总结,本人已拿大厂offer,欢迎star

原文链接:blog.ouyangsihai.cn >> 并发编程系列之CountDownLatch对战Cyclicbarrier


 上一篇
并发编程系列之Exchanger 并发编程系列之Exchanger
前言 上面我们介绍了信号量,再来说说交换者,这个东西用的不是很多,所以一般也不被经常关注,但是我们还是最好了解下,下面我将从什么是Exchanger以及如何使用Exchanger两个方面谈谈这个用于线程间协调的工具类。 什么是E
2021-04-05
下一篇 
并发编程系列之Fork,Join 并发编程系列之Fork,Join
前言 上节我们讲了阻塞队列,Java中的并发容器就算有了个基本的认识,今天我们来介绍一种线程工作模式,叫Fork/Join,他是JDK7之后提供的一个并行执行框架,主要的思想我觉得是分而治之,将一个大的任务分成多个小的任务并
2021-04-05