forked from uav/ard-uav-server
ard-device/src/main/java/com/ard/device/camera/service/impl/ArdCameraServiceImpl.java
@@ -373,4 +373,87 @@ redisService.setCacheObject(getCacheKey(camera.getId()), camera); } /** * 定时执行登录失败相机重新登录 */ @Scheduled(cron = "0/10 * * * * ?") public void failedReconnect() { try { log.debug("定时执行失败相机重连任务"); // 从数据库取全部设备 ArdCamera ardCamera = new ArdCamera(); ardCamera.setUseDataScope(false); List<ArdCamera> cameraList = this.selectArdCameraList(ardCamera); cameraList.stream() // 只处理登录失败或离线的设备 .filter(camera -> camera.getLoginId() == null || camera.getLoginId() < 0 || "0".equals(camera.getState())) .forEach(camera -> { log.info("尝试重新登录设备 {}:{}", camera.getIp(), camera.getPort()); cameraSDKService.login(camera); }); } catch (Exception ex) { log.error("定时执行失败相机重连任务异常：{}", ex.getMessage()); } } /** * 同步相机在线状态 */ @Scheduled(cron = "0/10 * * * * ?") public void syncCameraState() { try { log.debug("定时执行同步相机在线状态任务"); //调用查询方法 ArdCamera ardCamera = new ArdCamera(); ardCamera.setUseDataScope(false); List<ArdCamera> ardCameraList = this.selectArdCameraList(ardCamera); ardCameraList .forEach(camera -> { boolean onLineNew = cameraSDKService.isOnLine(camera); boolean onLineOld = "1".equals(camera.getState()); if (onLineNew != onLineOld) { if (onLineNew) { camera.setState("1"); } else { camera.setState("0"); } camera.setUpdateTime(DateUtils.getNowDate()); ardCameraMapper.updateArdCamera(camera); } }); } catch (Exception ex) { log.error("同步相机在线状态异常:{}", ex.getMessage()); } } /** * 同步相机通道 */ @Scheduled(cron = "0/10 * * * * ?") public void syncChannel() { try { log.debug("定时执行同步通道任务"); ArdCamera camera = new ArdCamera(); camera.setUseDataScope(false); List<ArdCamera> ardCameraList = this.selectArdCameraList(camera); if (!ardCameraList.isEmpty()) { ardCameraList .forEach(ardCamera -> { //SDK获取实际通道 List<ArdChannel> ardChannelListSDK = cameraSDKService.getChannels(ardCamera); //数据库获取通道 ArdChannel ardChannel = new ArdChannel(); ardChannel.setUseDataScope(false); ardChannel.setDeviceId(ardCamera.getId()); List<ArdChannel> ardChannelListDb = ardChannelService.selectArdChannelList(ardChannel); //同步通道 asyncChannel(ardChannelListDb, ardChannelListSDK); }); } } catch (Exception ex) { log.error("定时执行同步通道任务异常：{}", ex.getMessage()); } } } ard-device/src/main/java/com/ard/device/channel/service/impl/ArdChannelServiceImpl.java
@@ -33,7 +33,12 @@ @Resource private ArdChannelMapper ardChannelMapper; @Resource private RedisCache redisCache; @Resource private IArdVtduService ardVtduService; @Resource private MediamtxConfig mediamtxConfig; /** * 查询通道管理 * @@ -137,4 +142,110 @@ } return UserConstants.UNIQUE; } /** * 同步通道流媒体 */ @Scheduled(cron = "0/10 * * * * ?") public void syncVtdu() { log.debug("定时同步Vtdu任务"); try { ArdChannel ardChannel = new ArdChannel(); ardChannel.setUseDataScope(false); List<ArdChannel> channels = this.selectArdChannelList(ardChannel); if (channels == null || channels.isEmpty()) { // log.warn("没有获取到通道数据，退出同步"); return; } List<ArdVtdu> ardVtduList = ardVtduService.selectArdVtduList(new ArdVtdu()); Set<String> validNames = channels.stream() .map(channel -> channel.getDeviceId() + "_" + channel.getChanNo()) .collect(Collectors.toSet()); // 2. 合并操作：增、删、改 channels.forEach(channel -> { String name = channel.getDeviceId() + "_" + channel.getChanNo(); // 生成媒体表的 name String cameraId = channel.getDeviceId(); // cameraId 就是通道的 deviceId String rtspSourceUrl = getRtspSource(channel); String rtspUrl = "rtsp://" + mediamtxConfig.getPlayHost() + ":" + mediamtxConfig.getRtspPort() + "/" + name; // // 拼接 webrtcUrl String rtmpUrl = "rtmp://" + mediamtxConfig.getPlayHost() + ":" + mediamtxConfig.getRtmpPort() + "/" + name; // 拼接 // webrtcUrl String protocol = mediamtxConfig.isHttps() ? "https://" : "http://"; String webrtcUrl = protocol + mediamtxConfig.getPlayHost() + ":" + mediamtxConfig.getWebrtcPort() + "/" + name; // 拼接 webrtcUrl // 先检查媒体表是否已有此记录 Optional<ArdVtdu> existingMedia = ardVtduList.stream() .filter(media -> media.getName().equals(name)) // 查找是否已有对应的媒体记录 .findFirst(); if (existingMedia.isPresent()) { // 如果已有，更新现有记录 ArdVtdu vtdu = existingMedia.get(); // 更新前后内容相同则不进行更新 if (!vtdu.getRtspSource().equals(rtspSourceUrl) || !vtdu.getWebrtcUrl().equals(webrtcUrl) || !vtdu.getRtspUrl().equals(rtspUrl) || !vtdu.getRtmpUrl().equals(rtmpUrl)) { vtdu.setRtspUrl(rtspUrl); vtdu.setRtmpUrl(rtmpUrl); vtdu.setWebrtcUrl(webrtcUrl); ardVtduService.updateArdVtdu(vtdu); log.info("更新媒体记录：{}", vtdu.getName()); } } else { // 如果没有，插入新的记录 ArdVtdu vtdu = new ArdVtdu(); vtdu.setName(name); vtdu.setRtspSource(rtspSourceUrl); vtdu.setCameraId(cameraId); vtdu.setRtspUrl(rtspUrl); vtdu.setRtmpUrl(rtmpUrl); vtdu.setWebrtcUrl(webrtcUrl); ardVtduService.insertArdVtdu(vtdu); // 保存到媒体表 log.info("插入新媒体记录：{}", vtdu.getName()); } }); // 3. 删除操作：删除媒体表中多余的记录 List<String> invalidNames = ardVtduList.stream() .filter(vtdu -> !validNames.contains(vtdu.getName())) .filter(vtdu -> !"publisher".equalsIgnoreCase(vtdu.getRtspSource())) // 不是机场流 .map(ArdVtdu::getName) .collect(Collectors.toList()); if (!invalidNames.isEmpty()) { // 将 List<String> 转换为 String[] String[] namesArray = invalidNames.toArray(new String[0]); ardVtduService.deleteArdVtduByNames(namesArray); // 传递数组 log.info("删除不再存在的媒体记录：{}", Arrays.toString(namesArray)); } } catch (Exception e) { log.error("定时同步Vtdu任务异常:{}", e.getMessage()); } } // 根据通道信息获取rtsp源地址 private String getRtspSource(ArdChannel channel) { String rtspSource = ""; ArdCamera ardCamera = redisCache.getCacheObject(CacheConstants.CAMERA_LIST_KEY + channel.getDeviceId()); if (ardCamera != null) { switch (ardCamera.getFactory()) { case "1": rtspSource = "rtsp://" + ardCamera.getUsername() + ":" + ardCamera.getPassword() + "@" + ardCamera.getIp() + ":" + ardCamera.getRtspPort() + "/h264/ch" + channel.getChanNo() + "/main/av_stream"; break; case "2": rtspSource = "rtsp://" + ardCamera.getUsername() + ":" + ardCamera.getPassword() + "@" + ardCamera.getIp() + ":" + ardCamera.getRtspPort() + "/cam/realmonitor?channel=" + channel.getChanNo() + "&subtype=0"; break; } } return rtspSource; } } ard-quartz/src/main/java/com/ard/quartz/task/DeviceTask.java
File was deleted ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java
@@ -28,14 +28,14 @@ @Resource private ArdVtduMapper ardVtduMapper; @Resource private IArdMediaService mediaService; private IArdMediaService ardMediaService; @PostConstruct public void initMediaMTX() { //清空mediamtx流媒体数据 mediaService.clearAll(); ardMediaService.clearAll(); } /** @@ -69,7 +69,7 @@ @Override public int insertArdVtdu(ArdVtdu ardVtdu) { ardVtdu.setCreateTime(DateUtils.getNowDate()); ardVtdu = mediaService.addPath(ardVtdu); ardVtdu = ardMediaService.addPath(ardVtdu); return ardVtduMapper.insertArdVtdu(ardVtdu); } @@ -82,7 +82,7 @@ @Override public int updateArdVtdu(ArdVtdu ardVtdu) { ardVtdu.setUpdateTime(DateUtils.getNowDate()); ardVtdu = mediaService.editPath(ardVtdu); ardVtdu = ardMediaService.editPath(ardVtdu); ardVtdu.setUpdateTime(DateUtils.getNowDate()); return ardVtduMapper.updateArdVtdu(ardVtdu); } @@ -98,7 +98,7 @@ for (String name : names) { log.debug("流媒体【" + name + "】通道删除"); } mediaService.removePath(names); ardMediaService.removePath(names); return ardVtduMapper.deleteArdVtduByNames(names); } @@ -110,7 +110,7 @@ */ @Override public int deleteArdVtduByName(String name) { mediaService.removePath(name); ardMediaService.removePath(name); log.debug("流媒体【" + name + "】通道删除"); return ardVtduMapper.deleteArdVtduByName(name); } @@ -177,7 +177,7 @@ public void asyncMediaMTX(List<ArdVtdu> vtdus, List<String> names) { //需要更新的数据,参数顺序注意 sameList(vtdus, names).forEach(vtdu -> { Conf conf = mediaService.getPathConfig(vtdu.getName()); Conf conf = ardMediaService.getPathConfig(vtdu.getName()); if (conf == null || conf.getSource() == null) { return; } @@ -188,16 +188,32 @@ boolean isRtspSourceChanged = !Objects.equals(confRtspSource, currentRtspSource); // 如果 RTSP 源发生变化，则更新路径 if (isRtspSourceChanged) { mediaService.editPath(vtdu); ardMediaService.editPath(vtdu); } }); //需要删除的数据 diffListToDel(vtdus, names).forEach(name -> { mediaService.removePath(name); ardMediaService.removePath(name); }); //需要新增的数据 diffListToAdd(vtdus, names).forEach(vtdu -> { mediaService.addPath(vtdu); ardMediaService.addPath(vtdu); }); } /** * 同步mediamtx */ @Scheduled(cron = "0/10 * * * * ?") public void syncMediamtx() { log.debug("定时同步MediaMTX任务"); try { List<String> nameList = ardMediaService.getNameList(); List<ArdVtdu> ardVtduList = this.selectArdVtduList(new ArdVtdu()); asyncMediaMTX(ardVtduList, nameList); } catch (Exception e) { log.error("同步MediaMTX任务失败", e); } } } ard-work/src/main/java/com/ard/work/service/impl/ArdDroneInfoServiceImpl.java
@@ -307,57 +307,62 @@ */ @Scheduled(fixedRate = 5000) public void syncBoxInfo() { ArdServerInfo ardServerInfo = new ArdServerInfo(); ardServerInfo.setUseDataScope(false); List<ArdServerInfo> servers = ardServerInfoService.selectArdServerInfoList(ardServerInfo); log.debug("定时执行同步盒子属性任务"); try { ArdServerInfo ardServerInfo = new ArdServerInfo(); ardServerInfo.setUseDataScope(false); List<ArdServerInfo> servers = ardServerInfoService.selectArdServerInfoList(ardServerInfo); for (ArdServerInfo server : servers) { ArdDroneInfo droneInfo = new ArdDroneInfo(); droneInfo.setServerId(server.getId()); droneInfo.setUseDataScope(false); List<ArdDroneInfo> localList = selectArdDroneList(droneInfo); Map<String, ArdDroneInfo> localMap = localList.stream().collect(Collectors.toMap(ArdDroneInfo::getBoxSn, d -> d)); List<BoxInfoReceiver> boxList = tyService.getBoxList(server.getId()); // server 没有返回设备列表 → 全部离线 if (boxList == null || boxList.isEmpty()) { for (ArdDroneInfo local : localList) { if (local.getOnLine() != 0) { // 仅当状态变化才更新 local.setOnLine(0); ArdDroneInfoMapper.updateArdDrone(local); eventPublisher.publishEvent(new DeviceStatusEvent(local.getBoxSn(), false)); redisCache.deleteObject(CacheConstants.UAV_ONLINE_KEY + local.getBoxSn()); for (ArdServerInfo server : servers) { ArdDroneInfo droneInfo = new ArdDroneInfo(); droneInfo.setServerId(server.getId()); droneInfo.setUseDataScope(false); List<ArdDroneInfo> localList = selectArdDroneList(droneInfo); Map<String, ArdDroneInfo> localMap = localList.stream().collect(Collectors.toMap(ArdDroneInfo::getBoxSn, d -> d)); List<BoxInfoReceiver> boxList = tyService.getBoxList(server.getId()); // server 没有返回设备列表 → 全部离线 if (boxList == null || boxList.isEmpty()) { for (ArdDroneInfo local : localList) { if (local.getOnLine() != 0) { // 仅当状态变化才更新 local.setOnLine(0); ArdDroneInfoMapper.updateArdDrone(local); eventPublisher.publishEvent(new DeviceStatusEvent(local.getBoxSn(), false)); redisCache.deleteObject(CacheConstants.UAV_ONLINE_KEY + local.getBoxSn()); } } continue; } continue; } // server 返回设备列表 → 通过缓存判断状态 for (BoxInfoReceiver remote : boxList) { String boxSn = remote.getBoxSn(); String redisKey = CacheConstants.UAV_ONLINE_KEY + boxSn; boolean currentOnline = remote.getOnLine() == 1; // server 返回设备列表 → 通过缓存判断状态 for (BoxInfoReceiver remote : boxList) { String boxSn = remote.getBoxSn(); String redisKey = CacheConstants.UAV_ONLINE_KEY + boxSn; boolean currentOnline = remote.getOnLine() == 1; ArdDroneInfo local = localMap.get(boxSn); boolean dbOnline = local != null && local.getOnLine() == 1; boolean needUpdate = (dbOnline != currentOnline); // 只要远端设备在线，就刷新Redis缓存（无论状态是否变化），保证缓存一定存在 if (currentOnline) { redisCache.setCacheObject(redisKey, remote); } else { // 远端离线，删除Redis缓存 redisCache.deleteObject(redisKey); } // 如果状态变化才更新 if (needUpdate){ // 发布事件 eventPublisher.publishEvent(new DeviceStatusEvent(boxSn, currentOnline)); // 更新数据库 if (local != null) { local.setOnLine(currentOnline ? 1 : 0); updateArdDrone(local); ArdDroneInfo local = localMap.get(boxSn); boolean dbOnline = local != null && local.getOnLine() == 1; boolean needUpdate = (dbOnline != currentOnline); // 只要远端设备在线，就刷新Redis缓存（无论状态是否变化），保证缓存一定存在 if (currentOnline) { redisCache.setCacheObject(redisKey, remote); } else { // 远端离线，删除Redis缓存 redisCache.deleteObject(redisKey); } // 如果状态变化才更新 if (needUpdate) { // 发布事件 eventPublisher.publishEvent(new DeviceStatusEvent(boxSn, currentOnline)); // 更新数据库 if (local != null) { local.setOnLine(currentOnline ? 1 : 0); updateArdDrone(local); } } } } } catch (Exception e) { log.error("同步盒子属性任务异常", e); } }
@@ -48,38 +48,44 @@ */ @Scheduled(fixedRate = 1000) public void readTelemetryTask() { long now = System.currentTimeMillis(); UavTelemetryDataCache.forEach((boxSn, telemetryData) -> { // 1️⃣ 从状态缓存里取当前状态 UavState uavState = UavStateCache.get(boxSn); if (uavState == null) { log.warn("状态不存在，不记录"); return; // 状态不存在，不记录 } log.debug("定时任务：读取遥测数据"); try { long now = System.currentTimeMillis(); UavTelemetryDataCache.forEach((boxSn, telemetryData) -> { // 1️⃣ 从状态缓存里取当前状态 UavState uavState = UavStateCache.get(boxSn); if (uavState == null) { log.warn("状态不存在，不记录"); return; // 状态不存在，不记录 } // 2️⃣ 使用遥测自身时间戳判断是否还活着 long dataTs = telemetryData.getTimestamp(); // 设备时间 / 服务器接收时间 if (now - dataTs > 3000) { //log.warn("超过 3 秒没有新遥测 → 认为已离线或冻结，不记录：{}", new Date(dataTs)); return; // 超过 3 秒没有新遥测 → 认为已离线或冻结，不记录 } // 2️⃣ 使用遥测自身时间戳判断是否还活着 long dataTs = telemetryData.getTimestamp(); // 设备时间 / 服务器接收时间 if (now - dataTs > 3000) { //log.warn("超过 3 秒没有新遥测 → 认为已离线或冻结，不记录：{}", new Date(dataTs)); return; // 超过 3 秒没有新遥测 → 认为已离线或冻结，不记录 } int currentStatus = uavState.getFlightControllerState().getFlightStatus(); Integer lastStatus = lastFlightStatusMap.get(boxSn); int currentStatus = uavState.getFlightControllerState().getFlightStatus(); Integer lastStatus = lastFlightStatusMap.get(boxSn); // 仅在“飞行中 → 非飞行中”状态变化时 flush if (lastStatus != null && lastStatus == 2 && currentStatus != 2) { flushRemaining(boxSn); } lastFlightStatusMap.put(boxSn, currentStatus); log.debug("设备 {} 当前状态为：{}", boxSn, currentStatus); // 只有飞行中才添加缓存 if (currentStatus == 2) { log.warn("设备 {} 当前状态为飞行中，添加遥测数据到缓存", boxSn); UavFlightStateDTO dto = new UavFlightStateDTO(uavState, telemetryData, dataTs); addTelemetryData(dto); } }); // 仅在“飞行中 → 非飞行中”状态变化时 flush if (lastStatus != null && lastStatus == 2 && currentStatus != 2) { flushRemaining(boxSn); } lastFlightStatusMap.put(boxSn, currentStatus); log.debug("设备 {} 当前状态为：{}", boxSn, currentStatus); // 只有飞行中才添加缓存 if (currentStatus == 2) { log.warn("设备 {} 当前状态为飞行中，添加遥测数据到缓存", boxSn); UavFlightStateDTO dto = new UavFlightStateDTO(uavState, telemetryData, dataTs); addTelemetryData(dto); } }); } catch (Exception e) { log.error("定时任务：读取遥测数据异常", e); } } /** ard-work/src/main/java/com/ard/work/service/impl/ArdFlightTaskServiceImpl.java
@@ -55,12 +55,16 @@ */ @Scheduled(fixedDelay = 5000) public void syncAllFlightHistory() { List<String> serverIds = ardServerInfoService.getServerIdsList(); for (String serverId : serverIds) { // 获取上次同步的时间（即数据库中最大 startTime） Date lastSyncTime = getLastSyncTimeForServer(serverId); // 根据上次同步时间同步飞行历史 syncFlightHistory(serverId, lastSyncTime); try { List<String> serverIds = ardServerInfoService.getServerIdsList(); for (String serverId : serverIds) { // 获取上次同步的时间（即数据库中最大 startTime） Date lastSyncTime = getLastSyncTimeForServer(serverId); // 根据上次同步时间同步飞行历史 syncFlightHistory(serverId, lastSyncTime); } } catch (Exception e) { log.error("同步飞行历史失败", e); } }
File was renamed from ard-quartz/src/main/java/com/ard/quartz/task/MediaTask.java @@ -1,4 +1,4 @@ package com.ard.quartz.task; package com.ard.work.service.task; import com.ard.common.utils.StringUtils; import com.ard.common.utils.file.MimeTypeUtils; @@ -39,7 +39,7 @@ * * @author ard */ @Component("MediaTask") @Component @Slf4j public class MediaTask { @Resource @@ -55,7 +55,7 @@ private final AtomicBoolean runningVideo = new AtomicBoolean(false); private final AtomicBoolean runningPic = new AtomicBoolean(false); @Scheduled(cron = "0/10 * * * * ?") public void syncVideoMediaUrl() { log.warn("开始同步视频媒体"); if (!runningVideo.compareAndSet(false, true)) { @@ -81,7 +81,7 @@ runningVideo.set(false); } } @Scheduled(cron = "0/10 * * * * ?") public void syncPicMediaUrl() { log.warn("开始同步图片媒体"); if (!runningPic.compareAndSet(false, true)) {
