CountDownLatch使用示例demo

CountDownLatch使用示例demo

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * 模拟处理考试分值计算场景(包括答题卡扫描等)
 *
 * @author lyn
 * @date 2022/2/22 16:20
 */
@Slf4j
public class TestThreadPool {
    // TODO: 2022-11-18 注意 非必要不使用多线程
    // TODO: 2022-11-18 多线程并不一定比单线程快
    private static volatile AtomicInteger index = new AtomicInteger(0);
    private static ExecutorService executorService = Executors.newFixedThreadPool(5);

    public static void main(String[] args) {
        //准备基础数据
        final int totalSize = 10045;
        Random r = new Random();
        List<StuScore> list = new ArrayList<>();
        for (int i = 1; i <= totalSize; i++) {
            list.add(new StuScore(i, r.nextInt(55) + 40, r.nextInt(55) + 40, r.nextInt(55) + 40, 0));
        }
        long start = System.currentTimeMillis();

        //单线程计算 用时17450s
        //List<StuScore> result =singleJob(list);
        //多线程计算成绩 用时8573s
        List<StuScore> result = partJob(list);

        //取总分前五名展示
        List<StuScore> collect = result.stream().sorted(Comparator.comparing(StuScore::getTotalScore).reversed()).limit(5).collect(Collectors.toList());
        for (StuScore stuScore : collect) {
            log.info("学生{}", stuScore);
        }
        long cost = System.currentTimeMillis() - start;

        log.info("用时{}s", cost);
    }

    @SneakyThrows
    private static List<StuScore> partJob(List<StuScore> list) {
        //子集合的大小
        final int sonSize = 5000;
        //分割的份数
        int parts = (list.size() + sonSize - 1) / sonSize;
        //存放异步处理后结果数据
        List<StuScore> resultList = Collections.synchronizedList(new ArrayList<>());

        CountDownLatch countDownLatch = new CountDownLatch(parts);
        //分割集合
        List<List<StuScore>> partsList = IntStream.range(0, parts).boxed().parallel().map(i -> {
            int fromIndex = i * sonSize;
            long toIndex = sonSize;
            if (i + 1 == list.size()) {
                toIndex = list.size() - fromIndex;
            }
            return list.stream().skip(fromIndex).limit(toIndex).collect(Collectors.toList());
        }).collect(Collectors.toList());

        for (int i = 0; i < partsList.size(); i++) {
            List<StuScore> sonScoreList = partsList.get(i);
            //处理数据
            executorService.execute(new PartDisPose(i, sonScoreList, resultList, countDownLatch));
        }

        // 等待所有线程执行完毕
        countDownLatch.await();
        executorService.shutdown();
        return resultList;

    }

    /**
     * 分片处理内部类
     */
    private static class PartDisPose implements Runnable {

        private int count;
        private List<StuScore> sonList;
        private List<StuScore> resultList;
        private CountDownLatch countDownLatch;

        private PartDisPose(int count, List<StuScore> sonList, List<StuScore> resultList, CountDownLatch countDownLatch) {

            this.count = count;
            this.sonList = sonList;
            this.resultList = resultList;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            for (StuScore stu : sonList) {
                stu.setTotalScore(stu.getChinese() + stu.getEnglish() + stu.getMath());
                //模拟扫描答题卡等消耗的时间
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                index.incrementAndGet();
            }
            resultList.addAll(sonList);
            log.info("count: {},sonListSize: {}", count, sonList.size());
            countDownLatch.countDown();
        }
    }

    private static List<StuScore> singleJob(List<StuScore> list) {
        return list
                .stream()
                .peek(stu -> {
                    stu.setTotalScore(stu.getChinese() + stu.getEnglish() + stu.getMath());
                    //模拟扫描答题卡等消耗的时间
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                })
                .collect(Collectors.toList());
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
public class StuScore {
   private int id;
   private int math;
   private int english;
   private int chinese;
   private int totalScore;
}