#近3年发电量增加线程处理

hzims-service/hzims-scheduled/src/main/java/com/hnac/hzims/scheduled/service/operation/home/impl/RealTargetServiceImpl.java

@@ -10,7 +10,6 @@ import com.hnac.hzims.equipment.entity.EmParamEntity;
 import com.hnac.hzims.equipment.vo.EminfoAndEmParamVo;
 import com.hnac.hzims.hzimsweather.feign.IHeWeatherWeatherClient;
 import com.hnac.hzims.hzimsweather.vo.RainFallCountByMonthVo;
-import com.hnac.hzims.operational.config.vo.StationRealVo;
 import com.hnac.hzims.operational.fill.entity.RainfallEntity;
 import com.hnac.hzims.operational.home.wind.RainMon;
 import com.hnac.hzims.operational.main.constant.HomePageConstant;
@@ -40,7 +39,7 @@ import com.hnac.hzinfo.sdk.core.response.Result;
 import com.hnac.hzinfo.sdk.v5.device.DeviceDataClient;
 import com.hnac.hzinfo.sdk.v5.device.dto.ReductionDataDTO;
 import com.hnac.hzinfo.sdk.v5.device.vo.ReductionDataVO;
-import lombok.Getter;
+import io.undertow.util.CopyOnWriteMap;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.MapUtils;
@@ -113,8 +112,6 @@ public class RealTargetServiceImpl implements RealTargetService {
 	@Value("${hzims.operation.realIdKey}")
 	public String real_id_cofig_final;
 
-	// 创建线程池
-	ExecutorService pool = new ThreadPoolExecutor(8, 8, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), new ThreadFactoryBuilder().setNameFormat("load-power-data-pool-%d").build(), new ThreadPoolExecutor.AbortPolicy());
 
 	private final static String recent_year_power_data = "hzims:operation:key:power:data";
 	private final static String load_hydropower_unit_real_key = "hzims:operation:loadhydropowerunit:real:key";
@@ -126,6 +123,9 @@ public class RealTargetServiceImpl implements RealTargetService {
 	private final static String load_photovoltaic_real_key = "hzims:operation:photovoltaic:real:key";
 	private final static String load_photovoltaic_target_key = "hzims:operation:photovoltaic:target:key";
 
+	// 创建线程池
+	private static final  ExecutorService pool = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), new ThreadFactoryBuilder().setNameFormat("load-power-data-pool-%d").build() , new ThreadPoolExecutor.CallerRunsPolicy());
+
 	@Override
 	public void loadEmInfo(String param) {
 		List<StationEntity> stations =  stationService.list(Wrappers.<StationEntity>lambdaQuery()
@@ -1227,7 +1227,6 @@ public class RealTargetServiceImpl implements RealTargetService {
 	public void loadPowerData(String param, List<Integer> types, Integer serveType, int year) {
 		// 站点查询
 		List<StationEntity> stations = stationService.list(new LambdaQueryWrapper<StationEntity>() {{
-			eq(StationEntity::getDataOrigin,HomePageConstant.DATA_ORIGIN);
 			if (ObjectUtil.isNotEmpty(serveType)) {
 				eq(StationEntity::getServeType, serveType);
 			}
@@ -1248,14 +1247,13 @@ public class RealTargetServiceImpl implements RealTargetService {
 		calendar.add(Calendar.MONTH, -calendar.get(Calendar.MONTH));
 		String start = DateUtil.format(calendar.getTime(),DateUtil.PATTERN_DATE) + " 00:00:00";
 		// 存储数据map ：<站点id,<月份,发电量>>
-		Map<Long, Map<String, Float>> powerMap = new ConcurrentHashMap<>();
+		Map<Long, Map<String, Float>> powerMap = new CopyOnWriteMap<>();
 		// 将站点切割
 		int limit = countStep(stations.size());
-		List<List<StationEntity>> limits = Stream.iterate(0, n -> n + 1).limit(limit).parallel().map(a -> stations.stream().skip((long) a * 5).limit(5).parallel().collect(Collectors.toList())).collect(Collectors.toList());
-		// 线程数量
+		List<List<StationEntity>> limits = Stream.iterate(0, n -> n + 1).limit(limit).parallel().map(a -> stations.stream().skip((long) a * 3).limit(3).parallel().collect(Collectors.toList())).collect(Collectors.toList());
 		CountDownLatch countDownLatch = new CountDownLatch(limits.size());
-		pool.execute(()->{
-			for(List<StationEntity> item : limits){
+		for(List<StationEntity> item : limits){
+			pool.submit(()->{
 				item.forEach(station->{
 					// 站点设备集合
 					List<EminfoAndEmParamVo> stationDevices = devices.stream().filter(device -> device.getCreateDept().equals(station.getRefDept())).collect(Collectors.toList());
@@ -1266,20 +1264,22 @@ public class RealTargetServiceImpl implements RealTargetService {
 					powerMap.put(station.getId(),generateMap);
 				});
 				countDownLatch.countDown();
-			}
-		});
-		// 等待所有线程执行完成
+				log.error("pool_item_execute_complete: {}" ,item.stream().map(StationEntity::getName).collect(Collectors.toList()));
+			});
+
+		}
+		//所有模板数据获取完成后释放锁
 		try {
 			countDownLatch.await();
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 			Thread.currentThread().interrupt();
 		}
-		pool.shutdown();
 		redisTemplate.opsForValue().set(recent_year_power_data, powerMap);
 	}
 
 	@Override
+
 	public void loadPowerDataNew(List<Integer> types, Integer serveType, int year) {
 		// 站点查询
 		List<StationEntity> stationList = stationService.list(new LambdaQueryWrapper<StationEntity>() {{
@@ -1426,7 +1426,6 @@ public class RealTargetServiceImpl implements RealTargetService {
 	 */
 	private Map<String, Float> getGenerateYear(StationEntity station,List<EminfoAndEmParamVo> devices, String start, String end) {
 		List<PowerMonthVo> datas = new ArrayList<>();
-		long l = System.currentTimeMillis();
 		// 设备采集发电量
 		if(CollectionUtil.isNotEmpty(devices)){
 			devices.forEach(device->{
@@ -1447,17 +1446,11 @@ public class RealTargetServiceImpl implements RealTargetService {
 				}).collect(Collectors.toList()));
 			});
 		}
-		long l2 = System.currentTimeMillis();
-		log.info("getGenerateYear接口，"+station.getName()+"-设备采集发电量数据耗时："+(l2-l));
 		// 补充填报数据
 		datas.addAll(this.generateFill(station,start,end));
-		long l3 = System.currentTimeMillis();
-		log.info("getGenerateYear接口，"+station.getName()+"：补充填报数据"+(l3-l2));
 		if(CollectionUtil.isEmpty(datas)){
 			return null;
 		}
-		long l4 = System.currentTimeMillis();
-		log.info("getGenerateYear接口，"+station.getName()+"：接口总耗时"+(l4-l));
 		return datas.stream().collect(Collectors.toMap(PowerMonthVo::getStrMonth, PowerMonthVo::getPower, Float::sum));
 	}
 	/**