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.TMonitor; 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检测 */ 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(); //补充错误时间点 Query onlineQuery = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); Map mongoMap = mongoTemplate.find(onlineQuery, TMonitorResult.class).stream().collect(Collectors.toMap(TMonitorResult::getNo, Function.identity())); for (TMonitorResult result : monitorList) { TMonitorResult mongoData = mongoMap.get(result.getNo()); if(mongoData!=null){ result.setOffLineTimeStr(mongoData.getOffLineTimeStr()); } } 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); ex.printStackTrace(); } return null; // int checkTimes = 1; // int offLineTimes = 1; // Map map = (Map) redisTemplate.opsForHash().get(RedisConstant.ONLINE_KEY, monitor.getNo()); // if (!CollectionUtils.isEmpty(map)) { // checkTimes = (Integer) map.get("checkTimes") + 1; // offLineTimes = (Integer) map.get("offLineTimes"); // }else { // map = new HashMap<>(); // } // monitor.setPingOnline(Boolean.FALSE); // monitor.setCheckCount(checkTimes); // monitor.setOffLineCount(offLineTimes); // List offLineTime = monitor.getOffLineTime(); // if(CollectionUtils.isEmpty(offLineTime)) offLineTime = new ArrayList<>(); // offLineTime.add(new Date()); // monitor.setOffLineTime(offLineTime); // 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); // } // map.put("checkTimes", checkTimes); // map.put("offLineTimes", offLineTimes); // redisTemplate.opsForHash().put(RedisConstant.ONLINE_KEY, monitor.getNo(), map); // 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.error("部分任务未在指定时间内完成"); } catch (Exception e2){ log.error("数据收集异常"+e2); } dataList = futureList.stream() .map(CompletableFuture::join) .filter(Objects::nonNull) .collect(Collectors.toList()); Date now = new Date(); List offLineList = new ArrayList<>(); List onLineList = new ArrayList<>(); //查出数据库纯车辆或纯人脸设备 List serialNumbers = monitorMapper.selectCarOrFace().stream().map(TMonitor::getSerialNumber).collect(Collectors.toList()); dataList.forEach(item->{ //更新纯车辆或纯人脸的设备的在线状态(这部分设备不会拉流检测) if(item.getPingOnline() && !CollectionUtils.isEmpty(serialNumbers) && serialNumbers.contains(item.getNo())) { onLineList.add(item.getIp()); } else if(!item.getPingOnline()) { //筛选出ping离线的设备,更改数据库为离线 offLineList.add(item.getIp()); } }); if(!CollectionUtils.isEmpty(offLineList)) { monitorMapper.batchUpdateOnline(offLineList, now, ApiConstants.UY_OnlineSite_Offline); } if(!CollectionUtils.isEmpty(onLineList)) { monitorMapper.batchUpdateOnline(onLineList, now, ApiConstants.UY_OnlineSite_Online); } //存放到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 = new ArrayList<>(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("结束执行录像可用数据同步"); } }