美文网首页
CyclicBarrier原理

CyclicBarrier原理

作者: 光_93e5 | 来源:发表于2018-11-11 21:37 被阅读0次

    源码解析

    首先这个类有一个内部类

        //可以根据这个类内部的变量broken来判断这个CyclicBarrier是否能继续执行下去
        //当broken 被该为true说明中间发生了异常等。
        private static class Generation {
            boolean broken = false;
        }
    

    然后来看看这个类的构造方法

       // 设置一个常量parties记录这个CyclicBarrier状态和一个变量count 来记录当前状态值
       //当一个线程调用了await方法后,count就减1,直到为0时表示全部线程都已经达到要求了
       //如果传入的barrierAction不为null表示达到条件后执行barrierAction里的run方法
        public CyclicBarrier(int parties, Runnable barrierAction) {
            if (parties <= 0) throw new IllegalArgumentException();
            this.parties = parties;
            this.count = parties;
            this.barrierCommand = barrierAction;
        }
        //实际调用了上面的构造方法
        public CyclicBarrier(int parties) {
            this(parties, null);
        }
    

    主要的就是await方法,这个方法重写了两次

      //可以看出来一个是带有时间限制的,当达到一定时间还没达到要求,那么直 
      //接执行和被挂起的线程 
        public int await(long timeout, TimeUnit unit)
            throws InterruptedException,
                   BrokenBarrierException,
                   TimeoutException {
            return dowait(true, unit.toNanos(timeout));
        }
    
        public int await() throws InterruptedException, BrokenBarrierException {
            try {
                return dowait(false, 0L);
            } catch (TimeoutException toe) {
                throw new Error(toe); // cannot happen
            }
        }
    

    可以看到await中主要调用了dowait方法,我们来看看dowait方法中具体是怎么实现的

     //当timed为true时表示false时表示没有时间限制,为true表示经过一段时间还没有达到要求,那么就不管了,直接执行
    private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
                   TimeoutException {
            //先是上锁,
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                final Generation g = generation;
                //判断broken的值确认是否
                if (g.broken)
                    throw new BrokenBarrierException();
                //如果当前中断了,则抛出异常
                if (Thread.interrupted()) {
                   //先通知之前挂起的线程可以全部执行了,并且把count值至为初始值
                    breakBarrier();
                    throw new InterruptedException();
                }
                //把count值减1
                int index = --count;
                //如果count值为0了,表示达到已经条件
                if (index == 0) {  // tripped
                    //ranAction表示 barrierCommand方法还没执行,因为在执行run方法过程可能会异常中断
                    boolean ranAction = false;
                    try {
                        执行barrierCommand 中的run方法
                        final Runnable command = barrierCommand;
                        //判断是否为空
                        if (command != null)
                            command.run();
                        //执行到这里表示barrierCommand中的run方法正常执行完毕了
                        ranAction = true;
                         //通知其他挂起的线程可以继续执行了
                        nextGeneration();
                        return 0;
                    } finally {
                        //如果barrierCommand执行失败,通知之前挂起的线程可以全部执行了,并且把count值至为初始值
                        if (!ranAction)
                            breakBarrier();
                    }
                }
                //执行到这边说明还没有达到条件
                for (;;) {
                    try {
                        //timed为false直接挂起,
                        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();
            }
        }
    

    还是就是reset方法 是调用了breakBarrier和nextGeneration进行重置
    getNumberWaiting方法返回还需要差多少个能达到要求
    源码看完了,我们来看看例子是怎么使用的 ,附上代码

    public class Cyclic {
        public static void main(String[] args) throws InterruptedException {
            CyclicBarrier barrier  = new CyclicBarrier(4,new Thread(){
                public void run(){
                    System.out.println(Thread.currentThread().getName()+"其他线程已经执行完毕,执行当前线程");
                }
            });
            for(int i=0;i<4;i++) {
                new Writer(barrier).start();
                Thread.sleep(1000);
            }
    //      for(int i=0;i<4;i++) {
    //          new Writer(barrier).start();
     //         Thread.sleep(1000);
      //    }
        }
        static class Writer extends Thread{
            private CyclicBarrier cyclicBarrier;
            public Writer(CyclicBarrier cyclicBarrier) {
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+"正在写入数据..."+LocalTime.now());
                try {
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getName()+"写入数据完毕...等待其他线程"+LocalTime.now());
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }catch(BrokenBarrierException e){
                    e.printStackTrace();
                }
            }
        }
    }
    
    //执行结果
    Thread-1正在写入数据...21:24:34.405
    Thread-2正在写入数据...21:24:35.339
    Thread-1写入数据完毕...等待其他线程21:24:35.405
    Thread-2写入数据完毕...等待其他线程21:24:36.340
    Thread-3正在写入数据...21:24:36.341
    Thread-3写入数据完毕...等待其他线程21:24:37.341
    Thread-4正在写入数据...21:24:37.341
    Thread-4写入数据完毕...等待其他线程21:24:38.343
    Thread-4其他线程已经执行完毕,执行当前线程
    Thread-4继续执行任务
    Thread-1继续执行任务
    Thread-2继续执行任务
    Thread-3继续执行任务
    

    可以看到,先等待其中的任务都达到要求后,参数中的线程正确的执行了,然后其他线程继续执行,另外这个是可以重用的,如上代码把注释取消后,它会重新执行一遍。

    总结

    CyclicBarrier是可以让一组线程等待至某个状态之后在全部同时执行,而且这个是可以被重用的,当其中一个线程调用了await()方法后,那么当前线程就被挂起,直到达到某个条件后,最后一个完成条件的线程来执行barrierAction中的run方法,然后在执行其他挂起的线程

    相关文章

      网友评论

          本文标题:CyclicBarrier原理

          本文链接:https://www.haomeiwen.com/subject/ytzxfqtx.html