Callable接口以及CyclicBarrier实现list数据求和

旧城等待， 2022-01-11 06:29 321阅读 0赞

public class CountListIntegerWithCallable {
        public static void main(String[] args) {
            int threadNum = 10;
     
            ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
            List<Integer> list = new LinkedList<Integer>();
            for (int i = 0; i <100; ++i) {
                list.add(i);
            }
            int len = list.size() / threadNum;
            if (0 == len) {
                threadNum = list.size();
                len = 1;
            }
     
            List<Callable<Long>> callableList = new LinkedList<Callable<Long>>();
     
            for (int i = 0; i < threadNum; ++i) {
                List<Integer> subList = new LinkedList<Integer>();
                if (i == threadNum - 1) {//最后一个线程处理剩下所有任务
                    subList = list.subList(i * len, list.size());
                } else {
                    subList = list.subList(i * len, (i + 1) * len);
                }
                final List<Integer> finalSubList = subList;
                callableList.add(new Callable<Long>() {
                    public Long call() throws Exception {
                        long sum = 0;
                        for (int j = 0; j < finalSubList.size(); j++) {
                            sum += finalSubList.get(j);
                        }
                        System.out.println("分配给线程："+Thread.currentThread().getName()+" list元素部分计数和为："+sum);
                        return sum;
                    }
                });
            }
     
            try {
    //            这个不用自己控制等待，invokeAll执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表
                List<Future<Long>> futureList = executorService.invokeAll(callableList);
                for (Future<Long> future : futureList) {
                    System.out.println(future.get());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            } catch (Exception e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
     
        }
    }

public class CountSumWithCyclicBarrier {
        private CyclicBarrier cyclicBarrier;
        private long totalSum;
     
        private long getSum() {
            int threadNum = 10;
            ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
            List<Integer> list = new LinkedList<Integer>();
            for (int i = 0; i < 100; ++i) {
                list.add(i);
            }
            int len = list.size() / threadNum;
            if (0 == len) {
                threadNum = list.size();
                len = 1;
            }
            cyclicBarrier = new CyclicBarrier(threadNum + 1);
            List<Integer> subList = new LinkedList<Integer>();
            for (int i = 0; i < threadNum; ++i) {
                if (i == threadNum - 1) {//最后一个线程处理剩余list中数据
                    subList = list.subList(i * len, list.size());
                } else {
                    subList = list.subList(i * len, (i + 1) * len);
                }
                executorService.submit(new TaskDemo(subList, i));
     
            }
            try {
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            } catch (BrokenBarrierException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
            return totalSum;
     
        }
     
        public static void main(String[] args) {
            CountSumWithCyclicBarrier countSumWithCyclicBarrier = new CountSumWithCyclicBarrier();
            System.out.print(countSumWithCyclicBarrier.getSum());
        }
     
        class TaskDemo implements Runnable {
            private List<Integer> subList;
            private int i;
     
            public TaskDemo(List<Integer> subList, int i) {
                this.subList = subList;
                this.i = i;
            }
     
            public void run() {
                long sum = 0;
                for (int j = 0; j < subList.size(); ++j) {
                    sum += subList.get(j);
                }
     
                System.out.println("分配给线程：" + i + " " + Thread.currentThread().getName() + "那一部分List的整数和为Sum:" + sum);
                synchronized (TaskDemo.class) {
                    totalSum += sum;
                }
                //线程间相互等待，只有所有线程执行完后，才进行下面操作
     
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                }
     
            }
        }
    }

cyclicBarrier = new CyclicBarrier(threadNum + 1);  
**threadNum+1 加1会阻塞主线程**  
countDownLatch和CyclicBarrier区别：  
**countDownLatch是一个线程，等待其他几个线程完，才继续执行下面代码，例如多人赛跑  
CyclicBarrier与CountDownLatch类似，但是可以去完成好几个阶段的任务，比如上述三个人约好去地点A吃饭，到店吃玩饭后继续出发，到下一个目的地，全部到了之后再吃饭。   
此时CyclicBarrier可以完成循环等待。**

//3个人出发，每个人到店之后报告老板  
CyclicBarrier cb =new CyclicBarrier(3);  
for(...)\{  
   //每个人出发  
   go(cb);  
\}

//一个人的行程  
go(CyclicBarrier cb)\{  
   //到达地点1，等待其他人到齐  
   cb.await();  
   //都到了吃饭，然后再出发  
   eat then <a target=\_blank href="http://www.cxyclub.cn/Tag/continue.html" target="\_blank" style="outline-style: none; color: rgb(76, 120, 153);">continue</a> going;  
     
   //到达地点2，等待其他人到齐  
   cb.await();  
   //都到了吃饭  
   eat;  
\}

\---------------------   
  
原文：https://blog.csdn.net/liujn347/article/details/25500581