多线程处理大数据问题

绝地灬酷狼 2023-07-18 02:41 25阅读 0赞

# 前言 #

这两天公司要求也一个工具，对27类资源进行计算，看这个资源属于哪一个网格。  
一个资源的数据量大概是1万左右。然后网格的数据大概在5000.也就是说每种资源的每条数据都要个这5000多条的网格进行匹配，正常的话，一条数据找到对应的网格需要1分钟。但是要求整个一晚上跑完。  
所以就做了如下调整。  
每天定时任务执行一次。获取网格信息和27类资源信息。  
然后27类资源分别27个线程处理。然后每个资源中，再分子线程处理，每个子线程处理500条数据，这样1万的数据大概20个线程同时处理。  
整个程序跑起来的时候，大概线程数为600。tomcat 一般承受的并发线程数在1600 左右。所以这个方案可行。

# 主要代码 #

## GetResourceList ##

@Component
    public class GetResourceList {
    
        @Autowired
        GetTypeConfig typeConfig;
    
        @Autowired
        GetGridList GetGridList;
    
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
    
        public void process(){
            List<Map<String, Object>> typeConfigList=typeConfig.getConfigList();
            List<Map<String, Object>> gridList=GetGridList.getGridList();
            if(typeConfigList==null || typeConfigList.isEmpty() ||gridList==null || gridList.isEmpty()){
                return;
            }
            //new ResourceThread(typeConfigList.get(0),gridList,jdbcTemplate).start();
            for(Map<String, Object> map:typeConfigList){
                new ResourceThread(map,gridList,jdbcTemplate).start();
            }
        }
    }

typeConfig 是获取27类资源的类型和目标表，后面根据不同的类型获取资源列表。  
GetGridList 是网格数据列表。  
然后对这27类资源分别用27个线程处理。

## ResourceThread ##

@Override
        public void run() {
            for(int i=0;i<map.size();i++){
                String profession=map.get("专业").toString();
                String type=map.get("类别").toString();
                String tableName=map.get("目标表").toString();
                List<Map<String, Object>> resourceList=getListByType(profession,type);
                log.info("开始执行："+tableName+profession+type);
                log.info("resourceList："+resourceList.size());
                log.info("gridList："+list.size());
                deletDateFromTable(tableName);
                process(resourceList,list,tableName);
            }
        }
        public void process(List<Map<String, Object>> resourceList,List<Map<String, Object>> gridList,String tableName){
            if(resourceList==null || resourceList.isEmpty()){
                return;
            }
            int count=resourceList.size()/500 +1;
            log.info("线程数"+count);
            List<List<Map<String, Object>>> list=ListUtil.averageAssign(resourceList,count);
            for(List<Map<String, Object>> temp:list){
                new ResourceBranchThread(temp,gridList,jdbcTemplate,tableName).start();
            }
    
        }

这里可以看到，对每类资源，再用子线程去处理，每个子线程处理500 条数据。

## ResourceBranchThread ##

@Override
        public void run() {
            if(resourceList==null || resourceList.isEmpty()){
                return;
            }
            for(Map<String, Object> resourceMap:resourceList){
                SortResource sortResource=new SortResource(tableName,jdbcTemplate);
                sortResource.process(resourceMap,gridList);
            }
        }

## SortResource ##

@Slf4j
    public class SortResource {
    
        private String tableName;
        private JdbcTemplate jdbcTemplate;
    
        public SortResource(String tableName,JdbcTemplate jdbcTemplate){
            this.tableName=tableName;
            this.jdbcTemplate=jdbcTemplate;
        }
    
        public void process(Map<String, Object> map,List<Map<String, Object>> gridList){
            log.info("入库表："+tableName);
            double x=Double.parseDouble(map.get("经度").toString());
            double y=Double.parseDouble(map.get("纬度").toString());
            for(Map<String, Object> gridMap:gridList){
                String gridId=gridMap.get("ID").toString();
                List<List<String>> lists=lonLatInfoHandle(gridMap.get("经纬度信息").toString());
                //log.info("网格ID"+gridId);
                if(check(lists,x,y)){
                    insert(map,gridId);
                    break;
                }
            }
        }
    
        /**
         * 判断点是否在区域内
         * @param lists
         * @param x
         * @param y
         * @return
         */
        public boolean check(List<List<String>> lists,double x,double y){
            if(lists.isEmpty()){
                return false;
            }
            // 多多边型，有一个为true 就可以
            for(List<String> list:lists){
                if(checkHbracketsComma(list,x,y)){
                    return true;
                }
            }
            return false;
        }
    
    
        /**
         * 检测镂空的
         * @param list
         * @return
         */
        public boolean checkHbracketsComma(List<String> list,double x,double y){
            RegionExtern regExt = new RegionExtern(list.get(0));
            boolean flag=regExt.checkPointInRegion(x,y);
            if(list.size()>1 && flag){
                for(int i=1;i<list.size();i++){
                    RegionExtern regExtTemp = new RegionExtern(list.get(i));
                    if(regExt.checkPointInRegion(x,y)){
                        //如果存在镂空的区域内，则返回false;
                        flag=false;
                    }
                }
            }
            return flag;
        }
    
    
        /**
         * 经纬度信息范围处理
         * @param lonLatInfo
         */
        public static List<List<String>> lonLatInfoHandle(String lonLatInfo){
            List<List<String>> lists=new ArrayList<>();
            if(lonLatInfo==null || lonLatInfo.isEmpty()){
                 return lists;
            }
            //多多边型
            if(lonLatInfo.contains(ConstantPool.SEPARATORRIGHTRIGHTBRACKETSCOMMA)){
                String[] arr=lonLatInfo.split("\\)\\),");
                for(int i=0;i<arr.length;i++){
                    lists.add(rightHbracketsComma(arr[i]));
                }
            }else{
                List<String> list=rightHbracketsComma(lonLatInfo);
                lists.add(list);
            }
            return lists;
        }
    
        /**
         * 镂空或者多边形
         * @param lonLatInfo
         * @return
         */
        public static List<String> rightHbracketsComma(String lonLatInfo){
            List<String> list=new ArrayList<>();
            if(lonLatInfo.contains(ConstantPool.SEPARATORRIGHTBRACKETSCOMMA)){
                String[] arr=lonLatInfo.split("\\),");
                for(int i=0;i<arr.length;i++){
                    String temp=arr[i].replace(ConstantPool.MULTIPOLYGON,ConstantPool.SEPARATORNULL).replace(ConstantPool.POLYGON,ConstantPool.SEPARATORNULL).trim();
                    arr[i]=temp.replace(ConstantPool.SEPARATORLEFTBRACKETS,ConstantPool.SEPARATORNULL).replace(ConstantPool.SEPARATORRIGHTBRACKETS,ConstantPool.SEPARATORNULL).replace(ConstantPool.SEPARATORCOMMA,ConstantPool.SEPARATORBranch).replace(ConstantPool.SEPARATORSPACE,ConstantPool.SEPARATORCOMMA);
                }
                list= Arrays.asList(arr);
            }else {
                String temp=lonLatInfo.replace(ConstantPool.MULTIPOLYGON,ConstantPool.SEPARATORNULL).replace(ConstantPool.POLYGON,ConstantPool.SEPARATORNULL).trim();
                list.add(temp.replace(ConstantPool.SEPARATORLEFTBRACKETS,ConstantPool.SEPARATORNULL).replace(ConstantPool.SEPARATORRIGHTBRACKETS,ConstantPool.SEPARATORNULL).replace(ConstantPool.SEPARATORCOMMA,ConstantPool.SEPARATORBranch).replace(ConstantPool.SEPARATORSPACE,ConstantPool.SEPARATORCOMMA));
            }
            return list;
        }
    
    
        /**
         * 入库
         * @param map
         * @param gridId
         */
        public void insert(Map<String, Object> map,String gridId){
            String sql=String.format("insert [%s] select %s,%s,'%s','%s'",tableName,map.get("流水号").toString(),gridId,map.get("名称").toString(),map.get("详细地址").toString());
            log.info(sql);
            jdbcTemplate.execute(sql);
        }

# 将一个list均分成n个list #

/**
         * 将一个list均分成n个list,主要通过偏移量来实现的
         * @param source
         * @return
         */
        public static <T> List<List<T>> averageAssign(List<T> source,int n){
            List<List<T>> result=new ArrayList<List<T>>();
            int remaider=source.size()%n;  //(先计算出余数)
            int number=source.size()/n;  //然后是商
            int offset=0;//偏移量
            for(int i=0;i<n;i++){
                List<T> value=null;
                if(remaider>0){
                    value=source.subList(i*number+offset, (i+1)*number+offset+1);
                    remaider--;
                    offset++;
                }else{
                    value=source.subList(i*number+offset, (i+1)*number+offset);
                }
                result.add(value);
            }
            return result;
        }