zgyw.git - Gitblit

			@@ -3,39 +3,59 @@
			import com.alibaba.fastjson2.JSONObject;
			import com.mongodb.client.result.DeleteResult;
			import com.ycl.feign.UYClient;
			import com.ycl.platform.domain.param.UY.*;
			import com.ycl.platform.domain.entity.WorkOrder;
			import com.ycl.platform.domain.param.UY.ImageDetectionParam;
			import com.ycl.platform.domain.param.UY.MonitorQualifyParam;
			import com.ycl.platform.domain.param.UY.RecordMetaDSumParam;
			import com.ycl.platform.domain.param.UY.VideoOnlineParam;
			import com.ycl.platform.domain.result.SYS.TMonitorResult;
			import com.ycl.platform.domain.result.UY.ImageDetectionResult;
			import com.ycl.platform.domain.result.UY.MonitorQualifyResult;
			import com.ycl.platform.domain.result.UY.RecordMetaDSumResult;
			import com.ycl.platform.domain.result.UY.*;
			import com.ycl.platform.domain.result.UY.VideoOnlineResult;
			import com.ycl.platform.domain.vo.UpdateOnlineVO;
			import com.ycl.platform.mapper.TMonitorMapper;
			import com.ycl.platform.service.UYErrorTypeCheckService;
			import com.ycl.platform.service.WorkOrderService;
			import com.ycl.platform.service.YwPointService;
			import com.ycl.system.domain.SysConfig;
			import com.ycl.system.mapper.SysConfigMapper;
			import com.ycl.thread.OnlineCheckThread;
			import com.ycl.utils.CheckPointUtil;
			import com.ycl.utils.DateUtils;
			import constant.ApiConstants;
			import constant.RedisConstant;
			import enumeration.ErrorType;
			import enumeration.general.WorkOrderStatusEnum;
			import lombok.RequiredArgsConstructor;
			import lombok.extern.slf4j.Slf4j;
			import org.springframework.beans.factory.annotation.Value;
			import org.springframework.data.mongodb.core.MongoTemplate;
			import org.springframework.data.mongodb.core.query.Criteria;
			import org.springframework.data.mongodb.core.query.Query;
			import org.springframework.data.redis.core.RedisTemplate;
			import org.springframework.stereotype.Component;
			import org.springframework.util.CollectionUtils;

			import java.text.SimpleDateFormat;
			import java.util.Calendar;
			import java.util.Date;
			import java.util.List;
			import java.util.Objects;
			import java.util.*;
			import java.util.concurrent.*;
			import java.util.stream.Collectors;

			//优云对接数据任务
			@Slf4j
			@Component("UYTask")
			@RequiredArgsConstructor
			public class UYTask {

			private final RedisTemplate redisTemplate;
			private final MongoTemplate mongoTemplate;
			private final UYClient uyClient;
			private final UYErrorTypeCheckService uyErrorTypeCheckService;
			private final YwPointService pointService;

			private final TMonitorMapper monitorMapper;
			private final WorkOrderService workOrderService;
			private final SysConfigMapper sysConfigMapper;
			private final CheckPointUtil checkPointUtil;
			@Value("${youYun.tenantId}")
			private String tenantId;
			@Value("${youYun.apikey}")
			@@ -46,6 +66,14 @@
			private String email;
			@Value("${youYun.passwd}")
			private String passwd;

			private static final ExecutorService executorService = new ThreadPoolExecutor(16,
			128,
			5000,
			TimeUnit.SECONDS,
			new ArrayBlockingQueue<>(1000),
			new ThreadPoolExecutor.CallerRunsPolicy()
			);

			// 图像检测
			public void imageDetection() {
			@@ -143,7 +171,112 @@
			log.info("结束一机一档合格率数据同步");
			}

			//点位在线率
			/**
			* 点位在线检测
			*/
			public void pointOnline() throws ExecutionException, InterruptedException {
			log.info("开始检测点位在线");
			Integer times = 2;
			SysConfig config = new SysConfig();
			config.setConfigKey("DAY_OF_POINT_OUTLINE_TIMES");
			SysConfig sysConfig = sysConfigMapper.selectConfig(config);
			if (Objects.nonNull(sysConfig)) {
			times = Integer.valueOf(sysConfig.getConfigValue());
			} else {
			log.error("请配置离线次数，此次设置为默认值2");
			}
			// 先查出设备IP集合
			List<TMonitorResult> monitorList = monitorMapper.getDistinctIP();
			// List<Future<TMonitorResult>> futureList = new ArrayList<>(48);
			List<TMonitorResult> dataList = new ArrayList<>(48);
			// for (TMonitorResult monitor : monitorList) {
			// OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, times);
			// Future<TMonitorResult> future = executorService.submit(thread);
			// futureList.add(future);
			// }
			// for (Future<TMonitorResult> future : futureList) {
			// dataList.add(future.get()); // get方法会阻塞，知道拿到结果才继续执行for
			// }
			Integer time = times;
			List<CompletableFuture<TMonitorResult>> futureList = monitorList.stream()
			.map(monitor -> CompletableFuture.supplyAsync(() -> {
			OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, time);
			return thread.call(); // 假设 OnlineCheckThread 实现了 Callable 接口
			}, executorService)
			.orTimeout(30, TimeUnit.SECONDS)
			.exceptionally(ex -> {
			if (ex instanceof TimeoutException) {
			log.error("任务执行超时:");
			} else {
			log.error("任务执行异常:" + ex);
			}
			int checkTimes = 1;
			int offLineTimes = 1;
			Map<String, Object> map = (Map<String, Object>) redisTemplate.opsForHash().get(RedisConstant.ONLINE_KEY, monitor.getIp());
			if (!CollectionUtils.isEmpty(map)) {
			checkTimes = (Integer) map.get("checkTimes") + 1;
			offLineTimes = (Integer) map.get("offLineTimes");
			}
			monitor.setOnline(Boolean.FALSE);
			monitor.setCheckCount(checkTimes);
			monitor.setOffLineCount(offLineTimes);
			if (monitor.getOffLineCount() >= time) {
			WorkOrder workOrder = new WorkOrder();
			workOrder.setSerialNumber(monitor.getNo());
			List<String> errList = new ArrayList<>();
			errList.add(ErrorType.DEVICE_OFFLINE.getValue());
			workOrder.setErrorTypeList(errList);
			workOrder.setStatus(WorkOrderStatusEnum.DISTRIBUTED);
			monitor.setWorkOrder(workOrder);
			}
			return monitor; // 返回失败的结果
			}))
			.collect(Collectors.toList());

			// 等待所有任务完成，但不会无限期等待
			CompletableFuture<Void> allOf = CompletableFuture.allOf(
			futureList.toArray(new CompletableFuture[0])
			);
			try {
			allOf.get(35, TimeUnit.SECONDS); // 给予额外的5秒来收集结果
			} catch (TimeoutException e) {
			log.warn("部分任务未在指定时间内完成");
			}

			dataList = futureList.stream()
			.map(CompletableFuture::join)
			.filter(result -> result != null)
			.collect(Collectors.toList());
			// 更新point表的在线标识
			Date now = new Date();
			List<UpdateOnlineVO> willUpdateList = dataList.stream().map(item -> {
			UpdateOnlineVO vo = new UpdateOnlineVO();
			vo.setOnline(item.getOnline());
			vo.setIp(item.getIp());
			vo.setUpdateTime(now);
			return vo;
			}).collect(Collectors.toList());
			monitorMapper.updateOnline(willUpdateList);
			// 工单
			List<WorkOrder> workOrderList = dataList.stream()
			.filter(item -> Objects.nonNull(item.getWorkOrder()))
			.map(TMonitorResult::getWorkOrder)
			.collect(Collectors.toList());
			if (!CollectionUtils.isEmpty(workOrderList)) {
			workOrderService.innerAddWorkOrder(workOrderList);
			}
			//存放到mongo
			if (!CollectionUtils.isEmpty(dataList)) {
			//如果存在之前的数据先删除
			Query query = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));
			DeleteResult result = mongoTemplate.remove(query, TMonitorResult.class);
			//存放在mongo中
			mongoTemplate.insertAll(dataList);
			}
			log.info("点位在线监测完成");
			}

			//（弃用）点位在线率
			public void videoOnlineTask() {
			//视频图像质量
			log.info("开始执行点位在线数据同步");
			@@ -228,7 +361,6 @@
			// uyErrorTypeCheckService.recordMetaDSumCheck(mongoTemplate.find(query, RecordMetaDSumResult.class));
			log.info("结束执行录像可用数据同步");
			}



			}