本文共 8093 字，大约阅读时间需要 26 分钟。

一：简单介绍同步屏障CyclicBarrier.

      1.1 CyclicBarrier可以让一组线程到达一个屏障时被阻塞,直到最后一个线程到达屏障时,所有被屏障拦截的线程才会继续向下执行的.使用场景用于多线程计算数据.计算结果完成,插入同步屏障,阻塞等待.

      1.2 CyclicBarrier位于java.util.concurrent包下.线程内执行CyclicBarrier实例对象的await()方法后此线程已经到达了同步屏障了,在等待其他线程也到达屏障呐.

      1.3 CyclicBarrier的构造方法如下.

      传一个拦截线程数的参数.

public CyclicBarrier(int parties) {        this(parties, null);    }

     传一个拦截线程数的参数,提供一个线程任务,当线程到达屏障时,优先执行BarrierAction.

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

 二：基础使用演示

    2.1 两个线程的时候.一个主线程和一个自定义的子线程.

import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;/** * author: * date:  * time:  * description:CyclicBarrier * 一组线程到达一个屏障时被阻塞 */public class CyclicBarrierTest {    private static CyclicBarrier cyclicBarrier=new CyclicBarrier(2);        public static void main(String[] args) {        new Thread(new Runnable() {            @Override            public void run() {                try {                    cyclicBarrier.await();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }                System.out.println("Thread执行");            }        }).start();        try {            cyclicBarrier.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {            e.printStackTrace();        }        System.out.println("Main执行");    }}

 运行结果如下:(多次执行后,有一下两种结果,Java多线程的执行顺序是CPU随机调度的,和代码的物理位置的先后顺序无关!)

将CyclicBarrier的拦截线程数目设置为3的时候.(此时任然只有一个主线程和一个自定义的线程执行).只需如下修改,其他地方不变

private static CyclicBarrier cyclicBarrier=new CyclicBarrier(3);

执行结果如下：

这个时候主线程和自定义的线程都执行了await()方法,到达了屏障,但是没有第三个线程执行到屏障将不能放行此时阻塞的主线程个自定义线程了,只能等待了.

     2.2 现在执行CyclicBarrier的有优先任务的线程任务的.

import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;/** * author:  * date:  * time:  * description:CyclicBarrier * 一组线程到达一个屏障时被阻塞 */public class CyclicBarrierTest {    private static CyclicBarrier cyclicBarrier=new CyclicBarrier(2,new ImportAction());        public static void main(String[] args) {        new Thread(new Runnable() {            @Override            public void run() {                try {                    cyclicBarrier.await();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }                System.out.println("Thread执行");            }        }).start();        try {            cyclicBarrier.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {            e.printStackTrace();        }        System.out.println("Main执行");    }}class ImportAction implements Runnable{    @Override    public void run(){        System.out.println("执行ImportAction关键任务");    }}

执行结果如下:

(首先分析一下,应该是CPU随机调度其中一个线程,线程在其得到的时间片执行线程内的业务方法,然后执行了CyclicBarrier的await()方法,然后到达屏障阻塞了,在等待另外一个线程也到达屏障呐,但是现在有一个优先任务了,此时要执行优先线程任务了.).

执行完了,才会只需执行另外一个没有到达同步屏障的任务线程方法.

多次执行会有以下两种执行结果的.

下面的测试程序查看当前CyclicBarrier阻塞的线程数目.

import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;/** * author:  * date:  * time:  * description:CyclicBarrier * 一组线程到达一个屏障时被阻塞 */public class CyclicBarrierTest {    private static CyclicBarrier cyclicBarrier=new CyclicBarrier(2,new ImportAction());        public static void main(String[] args) {        new Thread(new Runnable() {            @Override            public void run() {                try {                    System.out.println(cyclicBarrier.getParties());                    System.out.println("此时拦截的线程屏障数："+ cyclicBarrier.getNumberWaiting());                    cyclicBarrier.await();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }                System.out.println("Thread执行");            }        }).start();        System.out.println("此时拦截的线程屏障数："+ cyclicBarrier.getNumberWaiting());        try {            cyclicBarrier.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {            e.printStackTrace();        }        System.out.println("Main执行");        System.out.println("此时拦截的线程屏障数："+ cyclicBarrier.getNumberWaiting());    }}class ImportAction implements Runnable{    @Override    public void run(){        System.out.println("执行ImportAction关键任务");    }}

运行结果：

三：实战演练

     3.1 场景描述：一个Excel保存了一个用户的多个账户这一年内银行消费情况,其中每个Sheet保存了该用户的一个账户一年的每一笔银行交易消费流水详细数目.现在要统计一下这个用户这一年平均每天银行消费账目数目.（假如该用户有四张不同银行的银行卡）.

    3.2 实现思路：为了计算效率,使用多线程分别统计出每个工作表的日均消费水平，都执行完之后就计算出了每个工作表的日均银行消费流水数目,然后使用BarrierAction计算出所有的账户日均银行消费流水账目,继而计算出该用户这一年日均银行消费流水账目了.

    3.3 设计可能出现的问题：

         ①：如果用单线程计算,由于数据计算量大,效率低.

         ②：采用多线程,但是不使用CyclicBarrier,这里是需要得到4个Sheet的日均流水才能得到整个用户一年的日均银行流水,需要等待都完成才可以的.不使用CyclicBarrier可能计算不方便,出现计算错误问题.

   3.4 编码实现.

          (使用多线程技术罗列：CyclicBarier,ConcurrentHashMap,FixedThreadPool,Runnable.

import java.util.Map;import java.util.concurrent.*;/** * author:  * date:  * time:  * description: */public class BankConsumeFlow implements  Runnable {    /** 获取当前PC的线程cpu数*/    private static final Integer CPU_NUMBER=Runtime.getRuntime().availableProcessors();    /** 设置同步屏障数目*/    private CyclicBarrier cyclicBarrier=new CyclicBarrier(4,this);    /** 保存每个工作表的银行消费日均流水*/    private ConcurrentHashMap
   
     sheetCount=new ConcurrentHashMap<>();    /** 核心计算方法*/    private void flowCount(){        Executor executor= Executors.newFixedThreadPool(CPU_NUMBER);        for(int i=0;i
    
      sheet:sheetCount.entrySet()){            result+=sheet.getValue();        }        // 得到最后计算结果        sheetCount.put("result", result);        System.out.println("该用户这一年的银行消费日均为："+result+"元");    }    public static void main(String[] args) {        BankConsumeFlow bankConsumeFlow=new BankConsumeFlow();        bankConsumeFlow.flowCount();    }}
    
   

每个Sheet的计算结果为100元,4个就是400元.

四：关键方法总结

        3.1 getParties()：返回同步屏障拦截的线程数目.     

public int getParties() {    return parties;}

        3.2 getNumberWaiting()：返回CyclicBarrier此时正在阻塞的线程数目.（0~getParties()-1）    

private int count;

public int getNumberWaiting() {    final ReentrantLock lock = this.lock;        lock.lock();    try {          return parties - count;        } finally {            lock.unlock();        }    }

  await调用了dowait()方法.

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

 使用等待特定时间后,就不会阻塞阻塞当前线程的.

public int await(long timeout, TimeUnit unit)        throws InterruptedException,               BrokenBarrierException,               TimeoutException {        return dowait(true, unit.toNanos(timeout));}

看下面这个初始化的构造方法如下.

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

 使用了ReentrantLock锁.返回的是阻塞线程总数减去线程计数的个数就是当前正在阻塞的.

      3.3 isBoken()方法用来了解阻塞的线程是否被中断了.

public boolean isBroken() {    final ReentrantLock lock = this.lock;        lock.lock();        try {            return generation.broken;        } finally {            lock.unlock();        } }

    3.4 reset()重新计算count,直接赋值为同步屏障初始化的阻塞的线程数.CountDownLatch的计数器只能使用一次.计算发生错误可以重置计算器,线程重新执行一次.

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

public void reset() {        final ReentrantLock lock = this.lock;        lock.lock();        try {            breakBarrier();   // break the current generation            nextGeneration(); // start a new generation        } finally {            lock.unlock();        } }

转载地址：http://tvll.baihongyu.com/