并发 | CyclicBarrier用于多线程计算数据并合并计算结果

CyclicBarrier用于多线程计算数据并合并计算结果

CyclicBarrier
CyclicBarrier让一组线程到达一个屏障（同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，
所有被屏障拦截的线程才会继续执行。
注：CyclicBarrier的计数器可以使用reset()方法重置

使用方式1：

package concurrency.cyclicbarrier;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest {
    static CyclicBarrier c = new CyclicBarrier(2);
    public static void main(String[] args) {
        new Thread(new Runnable(){
            @Override
            public void run(){
                try{
                    c.await();
                }catch(Exception e){
                    
                }
                System.out.println(1);
            }
        }).start();
        
        try{
            c.await();
        }catch(Exception e){
            
        }
        System.out.println(2);
    }
}

使用方式2：

package concurrency.cyclicbarrier;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest2 {
    static CyclicBarrier c = new CyclicBarrier(2, new A());
    
    public static void main(String[] args) {
        new Thread(new Runnable(){
            @Override
            public void run(){
                try{
                    c.await();
                }catch(Exception e){
                    
                }
                System.out.println(1);
            }
        }).start();
        
        try{
            c.await();
        }catch(Exception e){
            
        }
        System.out.println(2);
    }
    
    static class A implements Runnable{
        @Override
        public void run() {
            System.out.println(3);
        }
    }
}

应用场景： 一个Excel保存了用户所有银行流水，每个Sheet保存一个账户近一年的每笔银行流水，现在
需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完之后，得到每个sheet
的日均银行流水，最后，再用barrierAction用这些线程的计算结果，计算出整个Excel的日均银行流水。
代码：

package concurrency.cyclicbarrier;

import java.util.Map.Entry;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class BankWaterService implements Runnable{
    //创建4个屏障，处理完之后执行当前类的run方法
    private CyclicBarrier c = new CyclicBarrier(4, this);
    
    //假设只有4个sheet，所以只启动4个线程
    private Executor executor = Executors.newFixedThreadPool(4);
    
    //保存每个sheet计算出的银流结果
    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<String, Integer>();
    
    private void count(){
        for(int i = 0; i < 4; i++){
            executor.execute(new Runnable(){
                @Override
                public void run() {
                    //计算当前sheet的银流数据，计算代码省略
                    sheetBankWaterCount.put(Thread.currentThread().getName(), 1);
                    //银流计算完成，插入一个屏障
                    try{
                        c.await();
                    }catch(InterruptedException | BrokenBarrierException e){
                        e.printStackTrace();
                    }
                }
            });
        }
    }
    
    @Override
    public void run() {
        int result = 0;
        
        //汇总每个sheet计算出的结果
        for(Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()){
            result += sheet.getValue();
        }
        
        //将结果输出
        sheetBankWaterCount.put("result", result);
        System.out.println(result);
    }
    
    public static void main(String[] args){
        BankWaterService bankWaterCount = new BankWaterService();
        bankWaterCount.count();
    }
}