package com.ycl.task; import com.alibaba.fastjson2.JSONObject; import com.mongodb.client.result.DeleteResult; import com.ycl.feign.UYClient; 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.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.*; import java.util.concurrent.*; import java.util.function.Function; 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}") private String apikey; @Value("${youYun.accesskey}") private String accesskey; @Value("${youYun.email}") 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() { log.info("开始执行图像检测数据同步"); ImageDetectionParam param = new ImageDetectionParam(); param.setPageNum(ApiConstants.PageNo); param.setPageSize(ApiConstants.PageSize); param.setArealayerno(ApiConstants.AreaNo); JSONObject jsonObject = uyClient.imageDetection(param); if (jsonObject != null) { log.info("数据格式" + jsonObject); Integer statusCode = jsonObject.getInteger("statusCode"); if (ApiConstants.UYSuccessCode.equals(statusCode)) { JSONObject data = jsonObject.getJSONObject("data"); if (data != null) { List records = data.getList("records", ImageDetectionResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, ImageDetectionResult.class); records.stream().forEach(item -> { if (Objects.nonNull(item.getDeviceId())) { item.setNo(item.getDeviceId()); } }); pointService.setDeviceTagByGB(records); //存放在mongo中 mongoTemplate.insertAll(records); // 工单生成 uyErrorTypeCheckService.imageDetectionCheck(records); } else { log.error("图像监测数据为空{}", data); } } else { log.error("图像监测数据为空{}", jsonObject); } } else { log.error("图像监测请求失败{}", jsonObject); } } else { log.error("图像监测数据为空"); } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.imageDetectionCheck(mongoTemplate.find(query, ImageDetectionResult.class)); log.info("结束执行图像监测数据同步"); } //一机一档 public void monitorQualifyTask() { //一机一档合格率 // 一机一档注册率 // 档案考核比 log.info("开始执行一机一档合格率数据同步"); MonitorQualifyParam param = new MonitorQualifyParam(); param.setPageNum(ApiConstants.PageNo); param.setPageSize(ApiConstants.PageSize); JSONObject jsonObject = uyClient.monitorQualify(param); if (jsonObject != null) { String code = jsonObject.getString("code"); if (ApiConstants.UYSuccessCodeStr.equals(code)) { JSONObject data = jsonObject.getJSONObject("data"); if (data != null) { List records = data.getList("records", MonitorQualifyResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, MonitorQualifyResult.class); records.stream().forEach(item -> { if (Objects.nonNull(item.getSerialNumber())) { item.setNo(item.getSerialNumber().getShowValue()); } }); pointService.setDeviceTagByGB(records); pointService.setNew(records); //存放在mongo中 mongoTemplate.insertAll(records); // 工单生成 uyErrorTypeCheckService.monitorQualifyCheck(records); } else { log.error("一机一档合格率数据为空{}", data); } } } else { log.error("一机一档合格率数据为空{}", jsonObject); } } else { log.error("一机一档合格率数据为空"); } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.monitorQualifyCheck(mongoTemplate.find(query, MonitorQualifyResult.class)); log.info("结束一机一档合格率数据同步"); } /** * 点位在线PING检测 */ //TODO:点位在线(修改逻辑只针对工单) 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 monitorList = monitorMapper.getDistinctIP(); List dataList = new ArrayList<>(48); Integer time = times; List> futureList = monitorList.stream() .map(monitor -> CompletableFuture.supplyAsync(() -> { OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, time); return thread.call(); // 假设 OnlineCheckThread 实现了 Callable 接口 }, executorService) .orTimeout(60, TimeUnit.SECONDS) //出现异常 .exceptionally(ex -> { if (ex instanceof TimeoutException) { log.error("任务执行超时:"+monitor.getIp()); } else { log.error("任务执行异常:"+monitor.getIp() + ex); } int checkTimes = 1; int offLineTimes = 1; Map map = (Map) redisTemplate.opsForHash().get(RedisConstant.ONLINE_KEY, monitor.getIp()); if (!CollectionUtils.isEmpty(map)) { checkTimes = (Integer) map.get("checkTimes") + 1; offLineTimes = (Integer) map.get("offLineTimes"); } monitor.setPingOnline(Boolean.FALSE); monitor.setCheckCount(checkTimes); monitor.setOffLineCount(offLineTimes); if (monitor.getOffLineCount() >= time) { WorkOrder workOrder = new WorkOrder(); workOrder.setSerialNumber(monitor.getNo()); List 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 allOf = CompletableFuture.allOf( futureList.toArray(new CompletableFuture[0]) ); try { allOf.get(60, TimeUnit.SECONDS); // 给予额外的5秒来收集结果 } catch (TimeoutException e) { log.warn("部分任务未在指定时间内完成"); } catch (Exception e2){ log.error("数据收集异常"+e2); } dataList = futureList.stream() .map(CompletableFuture::join) .filter(Objects::nonNull) .collect(Collectors.toList()); //筛选出ping离线的设备,更改数据库为离线 Date now = new Date(); List willUpdateList = dataList.stream().filter(result->!result.getPingOnline()).map(item -> { UpdateOnlineVO vo = new UpdateOnlineVO(); vo.setOnline(ApiConstants.UY_OnlineSite_Offline); vo.setIp(item.getIp()); vo.setUpdateTime(now); return vo; }).collect(Collectors.toList()); monitorMapper.updateOnline(willUpdateList); //存放到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); } //工单(同一IP只生成一个工单) List workOrderList = dataList.stream() .filter(item -> Objects.nonNull(item.getWorkOrder())) .collect(Collectors.toMap( TMonitorResult::getIp, Function.identity(), (existing, replacement) -> existing // 如果遇到相同的 IP,保留第一个 TMonitorResult 对象 )) .values() .stream() .map(TMonitorResult::getWorkOrder) .collect(Collectors.toList()); if (!CollectionUtils.isEmpty(workOrderList)) { workOrderService.innerAddWorkOrder(workOrderList); } log.info("点位在线监测完成"); } //点位在线率(优云) public void videoOnlineTask() { //视频图像质量 log.info("开始执行点位在线数据同步"); VideoOnlineParam param = new VideoOnlineParam(); param.setPageNum(ApiConstants.PageNo); param.setPageSize(ApiConstants.PageSize); param.setArealayerno(ApiConstants.AreaNo); param.setStatus(ApiConstants.UY_OnlineStatus_All); param.setIcmpStatus(ApiConstants.UY_OnlineStatus_All); JSONObject jsonObject = uyClient.videoOnline(param); if (jsonObject != null) { Integer statusCode = jsonObject.getInteger("statusCode"); if (ApiConstants.UYSuccessCode.equals(statusCode)) { JSONObject data = jsonObject.getJSONObject("data"); if (data != null) { List records = data.getList("records", VideoOnlineResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, VideoOnlineResult.class); //打标签 records.forEach(item -> { if (Objects.nonNull(item.getDeviceId())) { item.setNo(item.getDeviceId()); } }); pointService.setDeviceTagByGB(records); //存放在mongo中 mongoTemplate.insertAll(records); //更新point表在线状态 Date now = new Date(); List willUpdateList = records.stream().map(item -> { UpdateOnlineVO vo = new UpdateOnlineVO(); vo.setOnline(item.getStatus()); vo.setIp(item.getIpAddr()); vo.setUpdateTime(now); return vo; }).collect(Collectors.toList()); monitorMapper.updateOnline(willUpdateList); //离线生成工单,一个ip只生成一个工单 List workOrders = (List) records.stream() .filter(item -> ApiConstants.UY_OnlineSite_Offline.equals(item.getStatus())) .collect(Collectors.toMap( VideoOnlineResult::getIpAddr, Function.identity(), (existing, replacement) -> existing // 如果遇到相同的 IP,保留第一个(existing) )).values(); uyErrorTypeCheckService.videoOnlineCheck(workOrders); } else { log.error("点位在线结果数据为空{}", data); } } else { log.error("点位在线结果数据为空{}", jsonObject); } } else { log.error("点位在线结果请求失败{}", jsonObject); } } else { log.error("点位在线结果数据为空"); } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.videoOnlineCheck(mongoTemplate.find(query, VideoOnlineResult.class)); log.info("结束执行点位在线数据同步"); } //录像可用 public void recordMetaDSumTask() { //录像可用率 log.info("开始执行录像可用数据同步"); RecordMetaDSumParam param = new RecordMetaDSumParam(); param.setTenantId(tenantId); Calendar instance = Calendar.getInstance(); instance.setTime(new Date()); instance.add(Calendar.DAY_OF_MONTH, -1); SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd"); String yesterday = format.format(instance.getTime()); param.setStatTime(yesterday); JSONObject jsonObject = uyClient.recordMetaDSumList(param); if (jsonObject != null) { if (ApiConstants.UYSuccessCodeStr.equals(jsonObject.getString("code"))) { List records = jsonObject.getList("data", RecordMetaDSumResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, RecordMetaDSumResult.class); records.stream().forEach(item -> { if (Objects.nonNull(item.getDeviceId())) { item.setNo(item.getDeviceId()); } }); //打标签 pointService.setDeviceTagByGB(records); //存放在mongo中 mongoTemplate.insertAll(records); // // 工单生成 // uyErrorTypeCheckService.recordMetaDSumCheck(records); } } else { log.error("录像可用数据为空{}", jsonObject); } } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.recordMetaDSumCheck(mongoTemplate.find(query, RecordMetaDSumResult.class)); log.info("结束执行录像可用数据同步"); } }