diff --git a/maibu-common/src/main/java/com/maibu/config/DeviceTask.java b/maibu-common/src/main/java/com/maibu/config/DeviceTask.java index 6975a3a..3e75d1c 100644 --- a/maibu-common/src/main/java/com/maibu/config/DeviceTask.java +++ b/maibu-common/src/main/java/com/maibu/config/DeviceTask.java @@ -102,6 +102,11 @@ public class DeviceTask { return builder(FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR); } + @Bean(FastBeeConstant.TASK.MAP_MESSAGE_PUSH) + public Executor mapMessagePush() { + return builder(FastBeeConstant.TASK.MAP_MESSAGE_PUSH); + } + @Bean(FastBeeConstant.TASK.DEVICE_TASK_HANDLER) public Executor deviceTaskHandler() { return builder(FastBeeConstant.TASK.DEVICE_TASK_HANDLER); diff --git a/maibu-common/src/main/java/com/maibu/constant/FastBeeConstant.java b/maibu-common/src/main/java/com/maibu/constant/FastBeeConstant.java index ddb6d4b..3e79804 100644 --- a/maibu-common/src/main/java/com/maibu/constant/FastBeeConstant.java +++ b/maibu-common/src/main/java/com/maibu/constant/FastBeeConstant.java @@ -105,6 +105,13 @@ public interface FastBeeConstant { * 设备错误 */ String DEVICE_HEARTBEAT_MONITOR = "deviceHeartBeatMonitor"; + + /** + * 设备错误 + */ + String MAP_MESSAGE_PUSH = "mapMessagePush"; + + /** * 设备任务处理线程 */ diff --git a/maibu-common/src/main/java/com/maibu/core/business/dto/DeviceMapPushDTO.java b/maibu-common/src/main/java/com/maibu/core/business/dto/DeviceMapPushDTO.java new file mode 100644 index 0000000..c620d22 --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/core/business/dto/DeviceMapPushDTO.java @@ -0,0 +1,18 @@ +package com.maibu.core.business.dto; + +import lombok.Data; + +@Data +public class DeviceMapPushDTO { + private String id; + private double lat; + private double lng; + private double y; //yaw + private double f; //fix_status + private double a; //altitude + private String s;//status 运行状态 + private Long t; //taskId 任务id + private double b; //电量 + private boolean ha = false;//hasAlarm 有无告警 + +} diff --git a/maibu-common/src/main/java/com/maibu/core/business/dto/WebMapMessageDTO.java b/maibu-common/src/main/java/com/maibu/core/business/dto/WebMapMessageDTO.java new file mode 100644 index 0000000..315532c --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/core/business/dto/WebMapMessageDTO.java @@ -0,0 +1,12 @@ +package com.maibu.core.business.dto; + +import com.maibu.core.enums.MesType; +import lombok.Data; + +import java.util.List; + +@Data +public class WebMapMessageDTO { + private List data; + private MesType type; +} diff --git a/maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterace.java b/maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterace.java deleted file mode 100644 index cb6f71c..0000000 --- a/maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterace.java +++ /dev/null @@ -1,7 +0,0 @@ -package com.maibu.core.business.inter; - -public interface DeviceWebSocketInterace { - - void sendWebsoketMes(String key, String data); - -} diff --git a/maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterface.java b/maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterface.java new file mode 100644 index 0000000..9eb4756 --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterface.java @@ -0,0 +1,10 @@ +package com.maibu.core.business.inter; + +public interface DeviceWebSocketInterface { + + void sendWebsocketMes(String key, String data); + + + void sendMapWebsocketMes(Long siteId,String data); + +} diff --git a/maibu-common/src/main/java/com/maibu/core/business/inter/WebsocketMesDispather.java b/maibu-common/src/main/java/com/maibu/core/business/inter/WebsocketMesDispather.java index e735127..1afb28c 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/inter/WebsocketMesDispather.java +++ b/maibu-common/src/main/java/com/maibu/core/business/inter/WebsocketMesDispather.java @@ -14,10 +14,19 @@ public class WebsocketMesDispather { private ApplicationContext applicationContext; public void dispather(String key, String data) { - Map beanMap = applicationContext.getBeansOfType(DeviceWebSocketInterace.class); + Map beanMap = applicationContext.getBeansOfType(DeviceWebSocketInterface.class); // 遍历 beanMap.values().forEach(x -> { - x.sendWebsoketMes(key, data); + x.sendWebsocketMes(key, data); + }); + } + + + public void mapMessageDispather(Long siteId, String data) { + Map beanMap = applicationContext.getBeansOfType(DeviceWebSocketInterface.class); + // 遍历 + beanMap.values().forEach(x -> { + x.sendMapWebsocketMes(siteId, data); }); } diff --git a/maibu-common/src/main/java/com/maibu/core/enums/MesType.java b/maibu-common/src/main/java/com/maibu/core/enums/MesType.java index 0579068..ebb1137 100644 --- a/maibu-common/src/main/java/com/maibu/core/enums/MesType.java +++ b/maibu-common/src/main/java/com/maibu/core/enums/MesType.java @@ -15,5 +15,6 @@ public enum MesType { detectionReport, // 障碍物消息 switchPermission, // 切换控制申请 switchResult,// 切换结果 - route_post;// 路径规划 点位推送 + route_post,// 路径规划 点位推送 + mapPush; } diff --git a/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java b/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java index eb9197c..4044ea8 100644 --- a/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java +++ b/maibu-common/src/main/java/com/maibu/memory/SiteMemory.java @@ -7,6 +7,7 @@ import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import com.maibu.core.business.*; +import com.maibu.core.business.dto.DeviceMapPushDTO; import com.maibu.mqtt.CustomMqttDeviceMonitor; import org.apache.commons.lang3.StringUtils; import org.eclipse.paho.client.mqttv3.MqttClient; @@ -56,6 +57,17 @@ public class SiteMemory { // 记录 设备上线后的运行数据 下线后清除并保存到数据库 public static ConcurrentHashMap deviceRunStatisticsMap = new ConcurrentHashMap<>(); + public static ConcurrentHashMap deviceMapPushMap = new ConcurrentHashMap<>(); + + public Map getAllDeviceMapPush() { + return deviceMapPushMap; + } + + public void saveMaPushDTO(String deviceId, DeviceMapPushDTO dto) { + deviceMapPushMap.put(deviceId, dto); + } + + public void saveRunStatistics(DeviceRunStatistics deviceRunStatistics) { if (deviceRunStatistics != null && !StringUtils.isEmpty(deviceRunStatistics.getDeviceId())) { deviceRunStatisticsMap.put(deviceRunStatistics.getDeviceId(), deviceRunStatistics); diff --git a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java index 3d91646..2efa4c0 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java @@ -3,6 +3,7 @@ package com.maibu.mqtt; import java.util.ArrayList; import java.util.List; +import com.maibu.core.business.dto.DeviceMapPushDTO; import org.springframework.util.CollectionUtils; import com.fasterxml.jackson.databind.JsonNode; @@ -160,7 +161,7 @@ public class HostMessageHandler implements CustomMqttMessageHandler { } public WebStatusMessageDTO createWebDeviceLocationMessage(HostLocationRelTimeDTO locationRelTimeDTO, - String deviceId) { + String deviceId) { WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO(); List transferData = new ArrayList<>(); DeviceStatusDetail lat = new DeviceStatusDetail(); @@ -217,6 +218,27 @@ public class HostMessageHandler implements CustomMqttMessageHandler { webStatusMessageDTO.setData(transferData); webStatusMessageDTO.setType(MesType.locationInfo); + + if (nettyDevice != null && nettyDevice.getDevice() != null) { + Long siteId = nettyDevice.getDevice().getSiteId(); + Long orgId = nettyDevice.getDevice().getOrgId(); + SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId); + if (siteMemory != null) { + DeviceMapPushDTO dto = new DeviceMapPushDTO(); + dto.setA(Double.parseDouble(altitude.getValue())); + dto.setId(deviceId); +// dto.setB(); +// dto.setHa(); + dto.setF(Double.parseDouble(fix_status.getValue())); + dto.setS(status.getValue()); + dto.setLat(Double.parseDouble(lat.getValue())); + dto.setLng(Double.parseDouble(lng.getValue())); + dto.setT(nettyDevice.getCurrentTaskId()); + dto.setY(Double.parseDouble(yaw.getValue())); + siteMemory.saveMaPushDTO(deviceId, dto); + } + } + return webStatusMessageDTO; } diff --git a/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java b/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java index 7e066a2..f2e0b43 100644 --- a/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java +++ b/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java @@ -42,6 +42,9 @@ public class InitThread implements ApplicationRunner { @Resource(name = FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR) private Executor deviceHeartBeatMonitorExecutor; + @Resource(name = FastBeeConstant.TASK.MAP_MESSAGE_PUSH) + private Executor mapMessagePushExecutor; + @Resource(name = FastBeeConstant.TASK.COMMON_DEVICE_MONITOR) private Executor commonDeviceMonitorExecutor; @@ -127,6 +130,11 @@ public class InitThread implements ApplicationRunner { GlobalMemory.initMqttDeviceMonitor(mqttHostUrl, mqttClientId, mqttUsername, mqttPassword); + // 地图监控推送 + mapMessagePushExecutor.execute(() -> { + deviceThreadService.mapMessagePush(); + }); + // 上位机设备心跳监控 deviceHeartBeatMonitorExecutor.execute(() -> { deviceThreadService.heartBeatMonitor(); diff --git a/maibu-netty-server/src/main/java/com/maibu/service/DevicePlanTaskMonitorService.java b/maibu-netty-server/src/main/java/com/maibu/service/DevicePlanTaskMonitorService.java index d4a9ed9..d013b92 100644 --- a/maibu-netty-server/src/main/java/com/maibu/service/DevicePlanTaskMonitorService.java +++ b/maibu-netty-server/src/main/java/com/maibu/service/DevicePlanTaskMonitorService.java @@ -229,9 +229,9 @@ public class DevicePlanTaskMonitorService { String deviceId = devicePlanTask.getDeviceId(); if (!StringUtils.isEmpty(deviceId) && deviceId.equals(key)) { devicePlanTask.setTaskStaus(DeviceTaskStaus.EXECUTING); - siteMemory.addDevicePlanExecute(devicePlanTask); device.setCurrentTaskId(devicePlanTask.getId()); device.setTask(devicePlanTask); + siteMemory.addDevicePlanExecute(devicePlanTask); siteMemory.removeDevicePlanPrepare(devicePlanTask); } } diff --git a/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java b/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java index c123ad6..99fb903 100644 --- a/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java +++ b/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java @@ -14,6 +14,10 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import com.maibu.core.business.dto.DeviceMapPushDTO; +import com.maibu.core.business.dto.WebMapMessageDTO; +import com.maibu.core.enums.*; +import com.maibu.utils.json.JsonUtils; import org.apache.commons.lang3.StringUtils; import org.checkerframework.checker.units.qual.s; import org.springframework.beans.factory.annotation.Autowired; @@ -35,11 +39,6 @@ import com.maibu.core.business.ErrorIdentificationStandard; import com.maibu.core.business.alarm_center.AlarmMessage; import com.maibu.core.business.device.NettyDevice; import com.maibu.core.business.inter.WebsocketMesDispather; -import com.maibu.core.enums.CompareEnum; -import com.maibu.core.enums.ConnectorStatus; -import com.maibu.core.enums.DeviceStatus; -import com.maibu.core.enums.ErrorSource; -import com.maibu.core.enums.RespondCode; import com.maibu.core.host.HeartBeatDTO; import com.maibu.core.host.LocationMessage; import com.maibu.core.redis.RedisCache; @@ -224,7 +223,7 @@ public class DeviceThreadService { } public List compareErrorStandard(Long siteId, Long orgId, DeviceRunningStatusHistory history, - List standards) { + List standards) { if (history == null || CollectionUtils.isEmpty(standards)) return null; List list = new ArrayList<>(); @@ -397,4 +396,36 @@ public class DeviceThreadService { }, 0, 500, TimeUnit.MILLISECONDS); } + + /** + * 监控心跳 + */ + public void mapMessagePush() { + executor.scheduleWithFixedDelay(() -> { + try { + List list = GlobalMemory.getAllSiteMemory(); + if (!CollectionUtils.isEmpty(list)) { + list.forEach(x -> { + Map map = x.getAllDeviceMapPush(); + if (!CollectionUtils.isEmpty(map)) { + List pushList = new ArrayList<>(map.values()); + WebMapMessageDTO dto = new WebMapMessageDTO(); + dto.setData(pushList); + dto.setType(MesType.mapPush); + try { + websocketMesDispather.mapMessageDispather(x.getSiteId(), JsonUtils.toJsonString(pushList)); + } catch (Exception e) { + log.error("实时地图数据websocket推送失败: siteId={}, error={}", x.getSiteId(), + e.getMessage(), e); + } + + } + }); + } + } catch (Exception e) { + log.error("执行监测心跳线程错误:{}", e.getMessage()); + } + }, 0, 100, TimeUnit.MILLISECONDS); + } + } diff --git a/maibu-web-middleware/src/main/java/com/maibu/dto/WebMessageBaseDTO.java b/maibu-web-middleware/src/main/java/com/maibu/dto/WebMessageBaseDTO.java index e1f4d6e..2d5e925 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/dto/WebMessageBaseDTO.java +++ b/maibu-web-middleware/src/main/java/com/maibu/dto/WebMessageBaseDTO.java @@ -8,6 +8,8 @@ import lombok.Data; public class WebMessageBaseDTO { private MesType type; private String userName; + private Long siteId; + private Long orgId; private String deviceId; private String token; private String message; diff --git a/maibu-web-middleware/src/main/java/com/maibu/memory/MiddleGlobalMemory.java b/maibu-web-middleware/src/main/java/com/maibu/memory/MiddleGlobalMemory.java index 6d8c8e3..e181f01 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/memory/MiddleGlobalMemory.java +++ b/maibu-web-middleware/src/main/java/com/maibu/memory/MiddleGlobalMemory.java @@ -9,8 +9,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; -import java.util.Map; -import java.util.Objects; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; @Component @@ -25,6 +24,8 @@ public class MiddleGlobalMemory { */ public static Vertx vertx; + //TODO 推送全局的时候需要 判断是不是单个的控制 + public static ConcurrentHashMap> mapPushMap = new ConcurrentHashMap<>(); // 存储所有在线WebSocket连接 // key: test:web:token public static final ConcurrentHashMap onlineSockets = new ConcurrentHashMap<>(); @@ -84,6 +85,7 @@ public class MiddleGlobalMemory { .findFirst().orElse(null); } + public static String findNettyClient(NettyClient nettyClient) { return nettyClientMap.entrySet().stream() .filter(entry -> Objects.equals(entry.getValue(), nettyClient)) @@ -91,5 +93,38 @@ public class MiddleGlobalMemory { .findFirst().orElse(null); } + public static void addMapSocket(Long siteId, String token, WebSocket webSocket) { + mapPushMap + .computeIfAbsent(siteId, k -> new ConcurrentHashMap<>()) + .put(token, webSocket); + } + + public static void removeByWebSocket(WebSocket webSocket) { + if (webSocket == null) { + return; + } + + mapPushMap.forEach((siteId, socketMap) -> { + socketMap.entrySet().removeIf(entry -> + entry.getValue() == webSocket + ); + + // siteId 下没有连接了,删除整个 siteId + if (socketMap.isEmpty()) { + mapPushMap.remove(siteId, socketMap); + } + }); + } + + public static List getMapSockets(Long siteId) { + Map map = mapPushMap.get(siteId); + + if (map == null || map.isEmpty()) { + return Collections.emptyList(); + } + + return new ArrayList<>(map.values()); + } + } diff --git a/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketHandler.java b/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketHandler.java index 5dc9ce3..3c65b2a 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketHandler.java +++ b/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketHandler.java @@ -64,6 +64,7 @@ public class WebsocketHandler extends WebSocketServer { releaseResource(key); logger.info("客户端断开连接: remoteAddress={}, code={}, reason={}, 当前在线数: {}", conn.getRemoteSocketAddress(), code, reason, MiddleGlobalMemory.onlineSockets.size()); } + MiddleGlobalMemory.removeByWebSocket(conn); } @Override @@ -77,22 +78,15 @@ public class WebsocketHandler extends WebSocketServer { String deviceId = webMessageBaseDTO.getDeviceId(); MesType type = webMessageBaseDTO.getType(); String key = userName + ":web" + ":" + token; - if (MesType.AUTH.equals(type)) { + if (MesType.mapPush.equals(type)) { + Long siteId = webMessageBaseDTO.getSiteId(); + MiddleGlobalMemory.addMapSocket(siteId, token, conn); + } else if (MesType.AUTH.equals(type)) { if (StringUtils.isEmpty(deviceId)) { return; } - //todo 先关闭原先的 -// WebSocket oldWebsocket = MiddleGlobalMemory.onlineSockets.get(key); NettyClient client = MiddleGlobalMemory.nettyClientMap.get(key); - -// if (oldWebsocket != null && oldWebsocket.isOpen()) { -// oldWebsocket.close(); -// } -// if (oldNettyClient != null && oldNettyClient.isConnected()) { -// oldNettyClient.disconnect(); -// } MiddleGlobalMemory.onlineSockets.put(key, conn); - //这里可以根据需要初始化Netty客户端 if (client == null) { client = new NettyClient(); } diff --git a/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketMesHandler.java b/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketMesHandler.java index 7154bb1..cb4aa6d 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketMesHandler.java +++ b/maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketMesHandler.java @@ -1,16 +1,19 @@ package com.maibu.websocket; -import com.maibu.core.business.inter.DeviceWebSocketInterace; +import com.maibu.core.business.inter.DeviceWebSocketInterface; import com.maibu.memory.MiddleGlobalMemory; import org.java_websocket.WebSocket; import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + +import java.util.List; @Component -public class WebsocketMesHandler implements DeviceWebSocketInterace { +public class WebsocketMesHandler implements DeviceWebSocketInterface { @Override - public void sendWebsoketMes(String key, String data) { + public void sendWebsocketMes(String key, String data) { MiddleGlobalMemory.onlineSockets.keySet().forEach(x -> { // key: test:web:token if (x.contains(key)) { @@ -21,4 +24,14 @@ public class WebsocketMesHandler implements DeviceWebSocketInterace { } + @Override + public void sendMapWebsocketMes(Long siteId, String data) { + List webSocketList = MiddleGlobalMemory.getMapSockets(siteId); + if (!CollectionUtils.isEmpty(webSocketList)) { + webSocketList.forEach(webSocket -> { + WebsocketHandler.sendMessageToClient(webSocket, data); + }); + } + } + }