java多线程读取、操作List集合

灰太狼 2022-05-24 05:21 448阅读 0赞

**大数据中，多线程分段list操作， demo如下**

/**
     * 每500条数据开启一条线程
     *
     * @Author: wpf
     * @Date: 16:05 2018/5/21
     * @Description: 
     * @param  * @param null  
     * @return   
     */
    private void threadExec2(List<SysCompany> list) throws ExecutionException, InterruptedException {
        // 开始时间
        long start = System.currentTimeMillis();
    
        LOG.info("list的大小：" + list.size());
    
        // 每500条数据开启一条线程
        int threadSize = 500;
        // 总数据条数
        int dataSize = list.size();
        // 线程数
        int threadNum = dataSize / threadSize + 1;
        // 定义标记,过滤threadNum为整数
        boolean special = dataSize % threadSize == 0;
        // 创建一个线程池
        ExecutorService exec = Executors.newFixedThreadPool(threadNum);
        // 定义一个任务集合
        List<Callable<SysCompany>> tasks = new ArrayList<Callable<SysCompany>>();
        Callable<SysCompany> task = null;
        List<SysCompany> cutList = null;
        // 确定每条线程的数据
        for (int i = 0; i < threadNum; i++) {
            if (i == threadNum - 1) {
                if (special) {
                    break;
                }
                cutList = list.subList(threadSize * i, dataSize);
            } else {
                cutList = list.subList(threadSize * i, threadSize * (i + 1));
            }
            // LOG.info("第" + (i + 1) + "组：" + cutList.toString());
            final List<SysCompany> listStr = cutList;
            task = new Callable<SysCompany>() {
    
                @Override
                public SysCompany call() throws Exception {
                    for(SysCompany c : listStr){
                        if(StringUtils.isNotEmpty(c.getUsername())){
                            LOG.info("timer：在执行中..." + c.getUsername());
                            // 业务...
                            sysCompanyService.getComCodeByInfoCompanyUser(c.getUsername());
                        }
                    }
                    return null;
                }
            };
            // 这里提交的任务容器列表和返回的Future列表存在顺序对应的关系
            tasks.add(task);
        }
        List<Future<SysCompany>> results = exec.invokeAll(tasks);
        /*for (Future<SysCompany> future : results) {
            SysCompany c = future.get();
        }*/
        // 关闭线程池
        exec.shutdown();
        LOG.info("线程任务执行结束");
        LOG.info("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
    }

/**
     * 固定开辟8线程
     *
     * @Author: wpf
     * @Date: 16:05 2018/5/21
     * @Description:
     * @param  * @param null
     * @return
     */
    private void threadExec1(List<SysCompany> list) throws ExecutionException, InterruptedException{
        // 开始时间
        long start = System.currentTimeMillis();
    
        // 总数据条数
        int listSize = list.size();
        LOG.info("总记录数为："+listSize);
        // 线程数
        int threadNum = 8;
    
        // 创建一个线程池
        ExecutorService exec = Executors.newFixedThreadPool(threadNum);
        // 定义一个任务集合
        List<Callable<Integer>> tasks = new ArrayList<Callable<Integer>>();
        Callable<Integer> task = null;
        List<SysCompany> cutList = null;
    
        // 均分算法
        Map<Integer,Long> map = allotOfAverage1(threadNum,listSize);
        int endValue = 0;
        int startValue = 0;
        int i = 0;
        for (Long v : map.values()) {
            endValue = (int)(startValue+v);
            cutList = list.subList(startValue, endValue);
            startValue += v;
            LOG.info("第" + ++i +"组大小为："+cutList.size());
            final List<SysCompany> listStr = cutList;
            task = new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    for(SysCompany c : listStr){
                        if(StringUtils.isNotEmpty(c.getUsername())){
    
                            // LOG.info("timer：在执行中..." + c.getUsername());
                            // 业务...
                            sysCompanyService.getComCodeByInfoCompanyUser(c.getUsername());
                        }
                    }
                    return 1;
                }
            };
            // 这里提交的任务容器列表和返回的Future列表存在顺序对应的关系
            tasks.add(task);
        }
    
        List<Future<Integer>> results = exec.invokeAll(tasks);
        for (Future<Integer> future : results) {
            LOG.info(future.get());
        }
        // 关闭线程池
        exec.shutdown();
        LOG.info("线程任务执行结束");
        LOG.info("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
    }

/*
     * 平均分配
     */
    public static Map<Integer,Long> allotOfAverage1(int threadNum, long listSize){
        Map<Integer,Long> allot=new HashMap<Integer,Long>(); //保存分配的信息
        for(int i=0;i<listSize;i++){
            int j=i%threadNum;
            if(allot.containsKey(j)){
                allot.put(j, allot.get(j) +1);
            }else{
                allot.put(j, 1L);
            }
        }
        return allot;
    }