栅栏 CyclicBarrier

  1 package cyclicbarrier;
  2 
  3 import java.util.concurrent.BrokenBarrierException;
  4 import java.util.concurrent.ConcurrentHashMap;
  5 import java.util.concurrent.CyclicBarrier;
  6 
  7 public class Total {   
  8     public static ConcurrentHashMap<String, Integer> result = new ConcurrentHashMap<String, Integer>();   
  9     public static void main(String[] args) { 
 10         
 11         TotalService totalService = new TotalServiceImpl();   
 12        CyclicBarrier barrier = new CyclicBarrier(5,   
 13                new TotalTask(totalService));   
 14         
 15         // 实际系统是查出所有省编码code的列表，然后循环，每个code生成一个线程。   
 16         new BillTask(new BillServiceImpl(), barrier, "北京").start();   
 17         new BillTask(new BillServiceImpl(), barrier, "上海").start();   
 18         new BillTask(new BillServiceImpl(), barrier, "广西").start();   
 19         new BillTask(new BillServiceImpl(), barrier, "四川").start();   
 20         new BillTask(new BillServiceImpl(), barrier, "黑龙江").start();
 21         
 22    }   
 23 }   
 24  
 25 /**  
 26  * 主任务：汇总任务  
 27  */  
 28 class TotalTask implements Runnable {   
 29    private TotalService totalService;   
 30   
 31     TotalTask(TotalService totalService) {   
 32         this.totalService = totalService;   
 33     }   
 34   
 35     public void run() {   
 36         // 读取内存中各省的数据汇总，过程略。   
 37         int totalCount = totalService.count();   
 38         Total.result.put("total", totalCount);
 39         System.out.println("=======================================");   
 40         System.out.println("开始全国汇总");  
 41         System.out.println("全国汇总结果:" +  Total.result.get("total"));
 42     }   
 43 }   
 44   
 45 /**  
 46  * 子任务：计费任务  
 47  */  
 48 class BillTask extends Thread {   
 49     // 计费服务   
 50     private BillService billService;   
 51     private CyclicBarrier barrier;   
 52     // 代码，按省代码分类，各省数据库独立。   
 53     private String code;   
 54  
 55     BillTask(BillService billService, CyclicBarrier barrier, String code) {   
 56         this.billService = billService;   
 57        this.barrier = barrier;   
 58        this.code = code;   
 59     }   
 60   
 61     public void run() {   
 62        System.out.println("开始计算--" + code + "省--数据！");   
 63         int ret = billService.bill(code); 
 64         Total.result.put(code, ret);
 65         // 把bill方法结果存入内存，如ConcurrentHashMap,vector等,代码略   
 66         System.out.println("结果:" + ret);
 67         System.out.println(code + "省已经计算完成,并通知汇总Service！");   
 68        try {   
 69             // 通知barrier已经完成   
 70             barrier.await();   
 71        } catch (InterruptedException e) {   
 72             e.printStackTrace();   
 73         } catch (BrokenBarrierException e) {   
 74             e.printStackTrace();   
 75        }   
 76     }   
 77 } 
 78 
 79 interface TotalService
 80 {
 81     public int count();
 82 }
 83 
 84 class TotalServiceImpl implements TotalService
 85 {
 86     @Override
 87     public int count()
 88     {
 89         int totalCount = 0;
 90         for(String key : Total.result.keySet())
 91         {
 92             totalCount += Total.result.get(key);
 93         }
 94         return totalCount;
 95     }
 96 }
 97 
 98 interface BillService
 99 {
100     public int bill(String code);
101 }
102 class BillServiceImpl implements BillService
103 {
104 
105     @Override
106     public int bill(String code)
107     {
108         return 1;
109     }
110 }