java 多线程并发查询mysql数据

小咪咪 2022-12-11 07:59 177阅读 0赞

> 程序猿学社的GitHub，欢迎**Star**  
> [github技术专题][github]  
> 本文已记录到github

### 文章目录 ###

*  前言
 *  需求
 *  思路
 *  代码

# 前言 #

> 用过mysql的朋友，对mysql性能应该有一定的感悟，数据量级在百万级别查询效率还行，到了千万以上查询性能直线下降。

# 需求 #

> 某数据采集平台，统计各个设备每个月的到报情况

*  建议使用hive，预处理或者跑任务的方式，因为mysql不适合做相关统计的功能。性能 hive>oracle>mysql

# 思路 #

*  使用多线程并发查询每一天的数据在合并。

# 代码 #

/** * 获取自动站站点汇总数据 * @param start * @param end */
    	public Map<Integer,ElementReportVo>  getZidongzhanSum(LocalDate start,LocalDate end) throws Exception{ 
    		//获取时间列表
    		List<String> betweenDates = ReportStatisUtil.getBetweenDate(start,end);
    		 //定义固定长度的线程池 防止线程过多
            ExecutorService executorService = Executors.newFixedThreadPool(15);
            
            List<ElementReportVo> result = new ArrayList<>();//返回结果
    
            List<Callable<List<ElementReportVo>>> tasks = new ArrayList<>();
    	    for (String str : betweenDates) { 
    	    	 Callable<List<ElementReportVo>> blockZidongzhan = getBlockZidongzhan(str+" 00:00:00",str+" 23:59:59");
    	         tasks.add(blockZidongzhan);
    		}
    	    //Future用于获取结果
            List<Future<List<ElementReportVo>>> futures=executorService.invokeAll(tasks);
            //处理线程返回结果
            if(futures!=null&&futures.size()>0){ 
                for (Future<List<ElementReportVo>> future:futures){     
                 result.addAll(future.get());
                }
            }
     
            executorService.shutdown();//关闭线程池
            
            Map<Integer, ElementReportVo> zidongzhanMap = getZidongzhanMap(result);
    		return zidongzhanMap;
    	}

/** * 查询自动站某一天的数据 * @param startTime * @param endTime * @return */
    	 private Callable<List<ElementReportVo>> getBlockZidongzhan(String startTime,String endTime){ 
    		 Callable<List<ElementReportVo>> callable = new Callable<List<ElementReportVo>>() { 
    	            public List<ElementReportVo> call() throws Exception { 
    	                HashMap<String,String> maps = new HashMap<>();
    	            	maps.put("startTime", startTime);
    	            	maps.put("endTime", endTime);
    	                List<ElementReportVo> list=stationQualityDetailInfoMapper.getZidongzhanLists(maps);
    	                return list;
    	            }
    	        };
    		 return callable; 
    	 };

/** * 获取自动站maps * @param list * @return */
    	public Map<Integer,ElementReportVo> getZidongzhanMap(List<ElementReportVo> list){ 
    		Map<Integer, List<ElementReportVo>> collect = list.stream().collect(Collectors.groupingBy(ElementReportVo::getStationInfoId));
    		Map<Integer,ElementReportVo> maps = new HashMap<Integer, ElementReportVo>();
    		for (Entry<Integer, List<ElementReportVo>> elEntry : collect.entrySet()) { 
    			//一个月的数据
    			List<ElementReportVo> elementReportVos = elEntry.getValue();
    			int spd = elementReportVos.stream().mapToInt(ElementReportVo::getSpd).sum();
    			int dir = elementReportVos.stream().mapToInt(ElementReportVo::getDir).sum();
    			int hum = elementReportVos.stream().mapToInt(ElementReportVo::getHum).sum();
    			int pa = elementReportVos.stream().mapToInt(ElementReportVo::getPa).sum();
    			int rain = elementReportVos.stream().mapToInt(ElementReportVo::getRain).sum();
    			int temp = elementReportVos.stream().mapToInt(ElementReportVo::getTemp).sum();
    			
    			ElementReportVo elementReportVo = maps.get(elEntry.getKey());
    			if(elementReportVo == null){ 
    				elementReportVo = new ElementReportVo();
    			}
    			elementReportVo.setStationInfoId(elEntry.getKey());
    			elementReportVo.setDir(dir);
    			elementReportVo.setHum(hum);
    			elementReportVo.setPa(pa);
    			elementReportVo.setRain(rain);
    			elementReportVo.setSpd(spd);
    			elementReportVo.setTemp(temp);
    			maps.put(elEntry.getKey(), elementReportVo);
    		}
    		return maps;
    	}

[github]: https://github.com/ITfqyd/cxyxs