多线程分段处理批量数据

本是古典何须时尚 2022-08-31 02:14 203阅读 0赞

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.*;
    import java.util.stream.Collectors;
    
    /**
     * @Description
     * @Author
     * @Date 2021/7/22 13:44
     */
    @Slf4j
    @Component
    public class ThreadHandle {
    
        // 最小分段大小
        private static final int MIN_SEGMENT_SITZ = 5;
        // 最大处理线程
        private static final int MAX_THREAD = 5;
        // springboot框架线程处理
        @Autowired
        private static Executor executor;
    
    
        public static void main(String[] args) {
            ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
            List<User> userList = getUserList();
            List<List<User>> partition = partition(userList);
            // 结果列表
            List<FutureTask<Integer>> resultList = new ArrayList<>();
            for (List<User> users : partition) {
                FutureTask<Integer> task = new FutureTask<>(new UserHandle(partition.indexOf(users), users));
                // FutureTask<Integer> task  = executor.execute(new FutureTask<>(new UserHandle(partition.indexOf(users), users)));
                newCachedThreadPool.execute(task);
                resultList.add(task);
            }
            resultList.forEach(e -> {
                try {
                    System.out.println("计算结束，第"+(e.get()+1));
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            });
        }
    
        public static List<User> getUserList() {
            List<User> userList = new ArrayList<>();
            for (int i = 0; i < 50000; i++) {
                User user = new User();
                user.setApplicationNum("" + i);
                user.setUserName("张"+i);
                userList.add(user);
            }
            return userList;
        }
    
    
        private static List<List<User>> partition(List<User> dbList) {
            List<List<User>> eventSegments = new ArrayList<List<User>>();
            // 1.先按申请单号编组
            Map<String, List<User>> eventGroup =
                    dbList.stream().collect(Collectors.groupingBy(User::getApplicationNum));
            // 2.分段组装
            // 2.1计算分段大小
            int segmentSize = dbList.size() / MAX_THREAD - 1;
            if (segmentSize < MIN_SEGMENT_SITZ) {
                // 每个线程至少处理业务笔数
                segmentSize = MIN_SEGMENT_SITZ;
            }
            // 2.2装填分段数据
            // 新建任务分段
            List<User> segment = new ArrayList<>();
            // 将分组事件循还装填任务分段
            for (Map.Entry<String, List<User>> eventEntry : eventGroup.entrySet()) {
                if (segment.size() < segmentSize) {
                    // 任务分段未填满时继续填充
                    segment.addAll(eventEntry.getValue());
                } else {
                    // 分段已装满则将该分段加入分段集合并开启新的分段
                    eventSegments.add(segment);
                    // 开启新的分段
                    segment = new ArrayList<>(eventEntry.getValue());
                }
            }
            // 将最后一个分段加入集合
            eventSegments.add(segment);
            // 3.返回分段集合
            return eventSegments;
        }
    
        static class UserHandle implements Callable<Integer> {
    
            private Integer seq;
            private List<User> userList;
    
            public UserHandle(Integer seq, List<User>  userList) {
                this.seq = seq;
                this.userList = userList;
            }
    
            @Override
            public Integer call() throws Exception {
                log.info("分段任务第{}段执行开始，共{}个长度。", seq + 1, userList.size());
                for (User user : userList) {
                    System.out.println("我是第【"+(seq+1)+"】段，我叫：" + user.getUserName());
                }
                log.info("分段任务第{}段执行结束，共{}个长度。", seq + 1, userList.size());
                return seq;
            }
        }
        
    }