diff --git a/maibu-common/src/main/java/com/maibu/core/business/DevicePlanTask.java b/maibu-common/src/main/java/com/maibu/core/business/DevicePlanTask.java index 96c9e60..2fd37a1 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/DevicePlanTask.java +++ b/maibu-common/src/main/java/com/maibu/core/business/DevicePlanTask.java @@ -40,6 +40,6 @@ public class DevicePlanTask extends OrgBaseDO { private LocalDateTime canExecuteTime; // 这个时间之前都不能执行 @TableField(exist = false) - private boolean taskAccepted = false; //任务被上位机接收 + private volatile boolean taskAccepted = false; //任务被上位机接收 } diff --git a/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java b/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java index 524c9f0..3b7b08a 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java +++ b/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java @@ -30,6 +30,10 @@ import java.util.ArrayList; import java.util.LinkedList; import java.util.List; import java.util.Queue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; @Slf4j @EqualsAndHashCode(callSuper = true) @@ -67,6 +71,9 @@ public class NettyDevice extends Connector { @JsonIgnore private WorkRecordMapper workRecordMapper = SpringUtils.getBean(WorkRecordMapper.class); + @JsonIgnore + private ScheduledExecutorService executor = SpringUtils.getBean("scheduledExecutorService"); + private static final String mqttClientId = "maibu-mqtt-client"; public NettyDevice(String connectorId) { @@ -80,7 +87,7 @@ public class NettyDevice extends Connector { } - public void status(Integer onlineStatus){ + public void status(Integer onlineStatus) { this.onlineStatus = onlineStatus; } @@ -318,9 +325,46 @@ public class NettyDevice extends Connector { if (executingTask) return; if (task != null) { + // startTime 在启动前记录 task.setStartTime(LocalDateTime.now()); - executingTask = true; - sendNavigationCommand(); + + AtomicInteger count = new AtomicInteger(0); + ScheduledFuture[] futureHolder = new ScheduledFuture[1]; + futureHolder[0] = executor.scheduleWithFixedDelay(() -> { + int current = count.incrementAndGet(); + try { + executingTask = true; + if (task.isTaskAccepted()) { + // 成功，停止后续执行 + futureHolder[0].cancel(false); + log.info("任务执行成功，停止重试，执行次数={}", current); + return; + } + sendNavigationCommand(); + // 达到最大次数 + if (current >= 10) { + futureHolder[0].cancel(false); + log.warn("任务执行达到最大次数，停止执行"); + executingTask = false; + task.setTaskStaus(DeviceTaskStaus.FAILED); + devicePlanTaskMapper.saveOrUpdate(task); + task = null; + } + + } catch (Exception e) { + log.error("任务执行异常，次数={}", current, e); + + // 异常是否算一次，可以根据你的业务决定 + if (current >= 10) { + futureHolder[0].cancel(false); + } + executingTask = false; + task.setTaskStaus(DeviceTaskStaus.FAILED); + devicePlanTaskMapper.saveOrUpdate(task); + task = null; + } + + }, 0, 3, TimeUnit.SECONDS); } } catch (Exception e) { throw new RuntimeException(e); diff --git a/maibu-common/src/main/java/com/maibu/core/domain/model/LoginUser.java b/maibu-common/src/main/java/com/maibu/core/domain/model/LoginUser.java index 161a1b2..308356f 100644 --- a/maibu-common/src/main/java/com/maibu/core/domain/model/LoginUser.java +++ b/maibu-common/src/main/java/com/maibu/core/domain/model/LoginUser.java @@ -48,6 +48,12 @@ public class LoginUser implements UserDetails { */ private String token; + + /** + * 用户唯一标识 用户控制设备 + */ + private String realToken; + /** * 登录时间 */ @@ -108,7 +114,7 @@ public class LoginUser implements UserDetails { this.permissions = permissions; } - public LoginUser(Long userId, Long deptId, Long orgId, Long siteId, Long roleId,String language, SysUser user, Set permissions) { + public LoginUser(Long userId, Long deptId, Long orgId, Long siteId, Long roleId, String language, SysUser user, Set permissions) { this.userId = userId; this.deptId = deptId; this.orgId = orgId; diff --git a/maibu-common/src/main/java/com/maibu/core/host/HostNavigationReplyDTO.java b/maibu-common/src/main/java/com/maibu/core/host/HostNavigationReplyDTO.java new file mode 100644 index 0000000..e459414 --- /dev/null +++ b/maibu-common/src/main/java/com/maibu/core/host/HostNavigationReplyDTO.java @@ -0,0 +1,19 @@ +package com.maibu.core.host; + + +import lombok.Data; +import lombok.EqualsAndHashCode; + +@EqualsAndHashCode(callSuper = true) +@Data +public class HostNavigationReplyDTO extends HostBase { + + private String task_id; //任务唯一编号 + private String command_id;//指令唯一编号 + private boolean accepted;//是否接收 + private String code;//任务动作 OK|INVALID|BUSY|ERROR + + private String message; + + +} diff --git a/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java b/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java index 79a78da..1df0561 100644 --- a/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java +++ b/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java @@ -109,7 +109,8 @@ public class GlobalMemory { MqttTopic.MOWER_WILDCARD_LOCATION,//位置 经纬度 MqttTopic.MOWER_WILDCARD_TASK_ROUTE, //路径回复 MqttTopic.MOWER_WILDCARD_ERROR,//错误 - MqttTopic.MOWER_WILDCARD_REALTIME //状态消息 + MqttTopic.MOWER_WILDCARD_REALTIME, //状态消息 + MqttTopic.MOWER_WILDCARD_TASK_TARGET_REPLY //路径规划下发回复 )); deviceConfig.setHandlerType(Constant.mqttHandleKey); //唯一的 用于处理上位机mqtt的 customMqttDeviceMonitor.registerDevice(deviceConfig); 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 89e4998..ce148d7 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java @@ -8,10 +8,7 @@ import com.maibu.core.business.device.NettyDevice; import com.maibu.core.business.dto.WebStatusMessageDTO; import com.maibu.core.business.inter.WebsocketMesDispather; import com.maibu.core.enums.MesType; -import com.maibu.core.host.HeartBeatDTO; -import com.maibu.core.host.HostLocationRelTimeDTO; -import com.maibu.core.host.HostTaskStatusDTO; -import com.maibu.core.host.HostVehicleRelTimeDTO; +import com.maibu.core.host.*; import com.maibu.memory.DeviceSessionManager; import com.maibu.memory.GlobalMemory; import com.maibu.utils.StringUtils; @@ -100,6 +97,9 @@ public class HostMessageHandler implements CustomMqttMessageHandler { case "control_reply": handleControlReply(deviceId, topic, payload); break; + case "route_reply": + handleRouteReply(deviceId, topic, payload); + break; default: log.debug("割草机消息: deviceId={}, topic={}, action={}", deviceId, topic, action); break; @@ -307,6 +307,23 @@ public class HostMessageHandler implements CustomMqttMessageHandler { } + /** + * 处理控制指令响应 + */ + private void handleRouteReply(String deviceId, String topic, String payload) { + log.debug("【路径任务下发响应】deviceId={}, payload={}", deviceId, payload); + + //todo 更新任务接收状态 + HostNavigationReplyDTO replyDTO = SpringUtils.getBean(HostNavigationReplyDTO.class); + NettyDevice device = deviceSessionManager.getDevice(deviceId); + if (replyDTO.isAccepted()) { + if (device != null && device.getTask() != null && String.valueOf(device.getTask().getId()).equals(replyDTO.getTask_id())) { + device.getTask().setTaskAccepted(true); + } + } + } + + /** * 从 Topic 中解析 deviceId * Topic 格式：mower/{deviceId}/... @@ -349,6 +366,7 @@ public class HostMessageHandler implements CustomMqttMessageHandler { if (remaining.equals("task/finish/post")) return "task_finish_post"; if (remaining.equals("action/config/reply")) return "config_reply"; if (remaining.equals("action/control/reply")) return "control_reply"; + if (remaining.equals("/task/target/reply")) return "route_reply"; return null; } diff --git a/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java index d27f192..d6c8160 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java @@ -94,6 +94,8 @@ public class MqttTopic { public static final String MOWER_WILDCARD_TASK_ROUTE = "mower/+/task/route/post"; + public static final String MOWER_WILDCARD_TASK_TARGET_REPLY = "mower/+/task/target/reply"; + public static final String MOWER_HEARTBEAT = "mower/+/event/heartbeat/post"; diff --git a/maibu-external/src/main/java/com/maibu/uav/service/UAVService.java b/maibu-external/src/main/java/com/maibu/uav/service/UAVService.java index 1d45162..a0a49a7 100644 --- a/maibu-external/src/main/java/com/maibu/uav/service/UAVService.java +++ b/maibu-external/src/main/java/com/maibu/uav/service/UAVService.java @@ -29,6 +29,7 @@ import org.springframework.util.CollectionUtils; import org.springframework.web.multipart.MultipartFile; import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; import java.io.IOException; import java.io.InputStream; import java.security.InvalidKeyException; @@ -58,6 +59,9 @@ public class UAVService { @Autowired private RedisCache redisCache; + @Autowired + private ScheduledExecutorService executor; + public static final Map latestActiveTime = new ConcurrentHashMap<>(); @@ -669,7 +673,6 @@ public class UAVService { @PostConstruct public void heartbeatMonitor() { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { Long nowTime = System.currentTimeMillis(); @@ -821,4 +824,5 @@ public class UAVService { return false; } + } diff --git a/maibu-framework/src/main/java/com/maibu/web/service/TokenService.java b/maibu-framework/src/main/java/com/maibu/web/service/TokenService.java index f47584f..3f2fdf8 100644 --- a/maibu-framework/src/main/java/com/maibu/web/service/TokenService.java +++ b/maibu-framework/src/main/java/com/maibu/web/service/TokenService.java @@ -170,8 +170,9 @@ public class TokenService { Map claims = new HashMap<>(); claims.put(Constants.LOGIN_USER_KEY, token); - return createToken(claims); - + String realToken = createToken(claims); + loginUser.setRealToken(realToken); + return realToken; } public int addToken(SysUser user, SysClient sysClient) { diff --git a/maibu-netty-server/src/main/java/com/maibu/controller/TransferDeviceController.java b/maibu-netty-server/src/main/java/com/maibu/controller/TransferDeviceController.java index 4773239..f4ee00c 100644 --- a/maibu-netty-server/src/main/java/com/maibu/controller/TransferDeviceController.java +++ b/maibu-netty-server/src/main/java/com/maibu/controller/TransferDeviceController.java @@ -66,11 +66,7 @@ public class TransferDeviceController extends BaseController { if (StringUtils.isEmpty(bindDTO.getPlatform())) { return AjaxResult.error("平台为空!"); } - LoginUser user = getLoginUser(); - if (user == null) { - return AjaxResult.error("当前token用户不存在!"); - } - return AjaxResult.success(transferDeviceService.switchDevice(bindDTO, user)); + return AjaxResult.success(transferDeviceService.switchDevice(bindDTO)); } /** @@ -87,8 +83,7 @@ public class TransferDeviceController extends BaseController { if (StringUtils.isEmpty(bindDTO.getPlatform())) { return AjaxResult.error("平台为空!"); } - LoginUser user = getLoginUser(); - return AjaxResult.success(transferDeviceService.remoteControl(bindDTO, user)); + return AjaxResult.success(transferDeviceService.remoteControl(bindDTO)); } catch (Exception e) { logger.error("获取设备控制权出错", e); return AjaxResult.error(); diff --git a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java index 2d00286..f6f22a0 100644 --- a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java +++ b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java @@ -192,6 +192,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { if (masterDevice == null) { // sessionManager.getControl(sessionId); handleMasterConnect(ctx, sessionId, null, ConnectorStatus.ACTIVE); + } else { + ctx.channel().attr(Constant.ATT_DEVICE_ID).set(sessionId); } } } @@ -250,7 +252,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { changeDTO.setTaskId(taskId); GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId), - JsonUtils.toJsonString(changeDTO)); + JsonUtils.toJsonString(changeDTO)); List controlMasters = sessionManager .getAllSlaveControl(deviceId); if (!CollectionUtils.isEmpty(controlMasters)) { @@ -273,7 +275,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { reportDTO.setEntity(device.getCurrentPoint()); reportDTO.setType(MesType.device_task_arrive_point); GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_ARRIVE_TOPIC, taskId), - JsonUtils.toJsonString(reportDTO)); + JsonUtils.toJsonString(reportDTO)); List controlMasters = sessionManager .getAllSlaveControl(deviceId); @@ -422,11 +424,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { private void handlePermissionRequest(DeviceRequestDTO requestDTO, byte[] bytes, ChannelHandlerContext ctx) { logger.info("转发控制请求 :{}", JsonUtils.toJsonString(requestDTO)); String slaveId = requestDTO.getDeviceId(); - List controlList = sessionManager.getSlaveControl(slaveId); String token = requestDTO.getToken(); -// String masterId = requestDTO.getUserId() + ":" + requestDTO.getPlatform(); - logger.info("转发控制请求 controlList :{}", JsonUtils.toJsonString(controlList)); - String master = sessionManager.getMaster(slaveId); if (!StringUtils.isEmpty(master)) { // 如果当前设备有控制 diff --git a/maibu-netty-server/src/main/java/com/maibu/netty/handler/HeartbeatHandler.java b/maibu-netty-server/src/main/java/com/maibu/netty/handler/HeartbeatHandler.java index 8c26224..824bbd8 100644 --- a/maibu-netty-server/src/main/java/com/maibu/netty/handler/HeartbeatHandler.java +++ b/maibu-netty-server/src/main/java/com/maibu/netty/handler/HeartbeatHandler.java @@ -31,6 +31,8 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter { private static final Logger logger = LoggerFactory.getLogger(HeartbeatHandler.class); + private static final long DEVICE_REGISTER_TIMEOUT_MS = 10_000; // 10秒注册超时 + @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { // 连接建立后启动定时任务，主动发心跳 @@ -39,6 +41,7 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter { } private void scheduleHeartbeat(ChannelHandlerContext ctx) { + final long startTime = System.currentTimeMillis(); ctx.executor().scheduleAtFixedRate(() -> { try { if (ctx.channel().isActive()) { @@ -47,6 +50,12 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter { ByteBuf heartbeat = Unpooled.wrappedBuffer(HEARTBEAT_PACKET); ctx.writeAndFlush(heartbeat); logger.debug("发送心跳包给客户端:{},deviceId:{}", ctx.channel().remoteAddress(), deviceId); + } else { + long elapsed = System.currentTimeMillis() - startTime; + if (elapsed > DEVICE_REGISTER_TIMEOUT_MS) { + logger.warn("设备超过{}ms未注册,断开连接:{}", DEVICE_REGISTER_TIMEOUT_MS, ctx.channel().remoteAddress()); + ctx.close(); + } } } } catch (Exception e) { @@ -55,6 +64,24 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter { }, 0, HEARTBEAT_INTERVAL, TimeUnit.SECONDS); } + +// private void scheduleHeartbeat(ChannelHandlerContext ctx) { +// ctx.executor().scheduleAtFixedRate(() -> { +// try { +// if (ctx.channel().isActive()) { +// String deviceId = ctx.channel().attr(Constant.ATT_DEVICE_ID).get(); +// if (!StringUtils.isEmpty(deviceId)) { +// ByteBuf heartbeat = Unpooled.wrappedBuffer(HEARTBEAT_PACKET); +// ctx.writeAndFlush(heartbeat); +// logger.debug("发送心跳包给客户端:{},deviceId:{}", ctx.channel().remoteAddress(), deviceId); +// } +// } +// } catch (Exception e) { +// logger.error("heartbeat error", e); +// } +// }, 0, HEARTBEAT_INTERVAL, TimeUnit.SECONDS); +// } + @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (msg instanceof ByteBuf) { 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 703d29c..d635f71 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 @@ -19,6 +19,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.util.CollectionUtils; +import javax.annotation.PreDestroy; import java.time.DayOfWeek; import java.time.LocalDate; import java.time.LocalDateTime; @@ -44,13 +45,12 @@ public class DevicePlanTaskMonitorService { @Autowired private WorkRecordMapper workRecordMapper; - + @Autowired + private ScheduledExecutorService executor; /** * 计划迟的任务生成机器执行任务 */ public void generatePlanTask() throws InterruptedException { - - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { Map> planMap = SiteMemory.devicePlanMap; @@ -204,7 +204,6 @@ public class DevicePlanTaskMonitorService { public void doLoop() throws InterruptedException { //prepare 进执行execute的逻辑 - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { List list = GlobalMemory.getAllSiteMemory(); @@ -250,7 +249,6 @@ public class DevicePlanTaskMonitorService { public void doExecute() { //prepare 进执行execute的逻辑 - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { List list = GlobalMemory.getAllSiteMemory(); @@ -270,5 +268,4 @@ public class DevicePlanTaskMonitorService { } }, 0, 1000, TimeUnit.MILLISECONDS); } - } 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 2c5fdce..5b0d5da 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 @@ -1,61 +1,48 @@ package com.maibu.service; -import java.io.IOException; -import java.lang.reflect.Field; -import java.math.BigDecimal; -import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Comparator; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; - -import com.maibu.constant.CommandConstant; -import com.maibu.constant.Constant; -import com.maibu.core.business.Device; -import com.maibu.core.enums.*; -import com.maibu.core.host.HeartBeatDTO; -import com.maibu.dto.DeviceStatusChangeDTO; -import com.maibu.memory.DeviceSessionManager; -import com.maibu.utils.CommandUtils; -import com.maibu.utils.json.JsonUtils; -import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelHandlerContext; -import io.netty.util.CharsetUtil; -import org.apache.commons.lang3.StringUtils; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.core.Constants; -import org.springframework.core.io.Resource; -import org.springframework.core.io.ResourceLoader; -import org.springframework.stereotype.Service; -import org.springframework.util.CollectionUtils; - +import cn.hutool.core.io.IORuntimeException; +import cn.hutool.core.io.IoUtil; import com.alibaba.fastjson2.JSON; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.maibu.common.NettyCacheKey; +import com.maibu.constant.Constant; +import com.maibu.core.business.Device; import com.maibu.core.business.DeviceRunningStatusHistory; import com.maibu.core.business.DeviceStatusRecordDTO; 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.*; +import com.maibu.core.host.HeartBeatDTO; import com.maibu.core.redis.RedisCache; import com.maibu.dto.DeviceErrorPushDTO; import com.maibu.mapper.ErrorIdentificationStandardMapper; +import com.maibu.memory.DeviceSessionManager; import com.maibu.memory.GlobalMemory; import com.maibu.memory.SiteMemory; import com.maibu.mqtt.MqttTopic; import com.maibu.utils.CompareUtils; - -import cn.hutool.core.io.IORuntimeException; -import cn.hutool.core.io.IoUtil; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.core.io.Resource; +import org.springframework.core.io.ResourceLoader; +import org.springframework.stereotype.Service; +import org.springframework.util.CollectionUtils; + +import javax.annotation.PreDestroy; +import java.io.IOException; +import java.lang.reflect.Field; +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; @Slf4j @Service @@ -85,13 +72,14 @@ public class DeviceThreadService { @Value("${mqtt.client-id}") private String mqttClientId; + @Autowired + private ScheduledExecutorService executor; + public void deviceErrorMonitor() throws InterruptedException, IOException { initStandard(); // TODO fix 后续是否分多个线程 一个site一个线程 // 有还需要更新接受故障方式。 - - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { String key = NettyCacheKey.deviceRunningStatusKey; @@ -270,8 +258,6 @@ public class DeviceThreadService { * 监控心跳 */ public void heartBeatMonitor() { - - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { Long nowTime = System.currentTimeMillis(); diff --git a/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java b/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java index 23bc24c..9e78e14 100644 --- a/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java +++ b/maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java @@ -220,7 +220,7 @@ public class TransferDeviceService { // return false; // } - public synchronized boolean switchDevice(DeviceBindDTO bindDTO, LoginUser user) { + public synchronized boolean switchDevice(DeviceBindDTO bindDTO) { try { String slaveId = bindDTO.getDeviceId(); // String platform = bindDTO.getPlatform(); @@ -270,9 +270,9 @@ public class TransferDeviceService { } - public DeviceControlDTO remoteControl(DeviceBindDTO bindDTO, LoginUser user) { + public DeviceControlDTO remoteControl(DeviceBindDTO bindDTO) { try { - String token = user.getToken(); + String token = bindDTO.getToken(); String masterToken = sessionManager.getMaster(bindDTO.getDeviceId()); DeviceControlDTO dto = new DeviceControlDTO(); if (!StringUtils.isEmpty(token) && token.equals(masterToken)) { diff --git a/maibu-service/maibu-system-service/src/main/java/com/maibu/service/impl/IotDeviceCommonService.java b/maibu-service/maibu-system-service/src/main/java/com/maibu/service/impl/IotDeviceCommonService.java index 72985a6..5556526 100644 --- a/maibu-service/maibu-system-service/src/main/java/com/maibu/service/impl/IotDeviceCommonService.java +++ b/maibu-service/maibu-system-service/src/main/java/com/maibu/service/impl/IotDeviceCommonService.java @@ -4,21 +4,20 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.maibu.constant.Constant; import com.maibu.core.business.DeviceModelPropertyMapping; -import com.maibu.core.business.DevicePlanTask; import com.maibu.core.business.IoTCommonDevice; import com.maibu.core.business.Product; -import com.maibu.core.business.device.NettyDevice; import com.maibu.core.business.dto.DeviceConfigDTO; import com.maibu.core.enums.ConnectType; import com.maibu.mapper.IotDeviceCommonMapper; import com.maibu.mapper.SysDeviceModelPropertyMappingMapper; import com.maibu.memory.GlobalMemory; import com.maibu.memory.SiteMemory; -import com.maibu.mqtt.*; +import com.maibu.mqtt.CustomMqttDeviceConfig; +import com.maibu.mqtt.CustomMqttDeviceMonitor; +import com.maibu.mqtt.GenericPropertyMessageHandler; +import com.maibu.mqtt.PropertyMappingConfig; import com.maibu.utils.SecurityUtils; import lombok.extern.slf4j.Slf4j; -import org.eclipse.paho.client.mqttv3.MqttClient; -import org.eclipse.paho.client.mqttv3.MqttException; import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -26,14 +25,13 @@ import org.springframework.transaction.annotation.Transactional; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +import javax.annotation.PreDestroy; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.HashMap; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -53,6 +51,9 @@ public class IotDeviceCommonService { @Autowired private SysDeviceModelPropertyMappingMapper sysDeviceModelPropertyMappingMapper; + @Autowired + private ScheduledExecutorService executor; + public void save(IoTCommonDevice ioTCommonDevice) throws Exception { iotDeviceCommonMapper.saveOrUpdate(ioTCommonDevice); SiteMemory siteMemory = GlobalMemory.getSiteMemory(ioTCommonDevice.getOrgId(), ioTCommonDevice.getSiteId()); @@ -269,8 +270,6 @@ public class IotDeviceCommonService { public static Map> mappingCache = new ConcurrentHashMap<>(); public void commonDeviceMonitor() { - - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleWithFixedDelay(() -> { try { CustomMqttDeviceMonitor monitor = GlobalMemory.customMqttDeviceMonitor; diff --git a/maibu-web-middleware/src/main/java/com/maibu/controller/WebMiddlewareController.java b/maibu-web-middleware/src/main/java/com/maibu/controller/WebMiddlewareController.java index 9eec11b..769ab4a 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/controller/WebMiddlewareController.java +++ b/maibu-web-middleware/src/main/java/com/maibu/controller/WebMiddlewareController.java @@ -62,10 +62,11 @@ public class WebMiddlewareController extends BaseController { List nettyClients = new ArrayList<>(); MiddleGlobalMemory.nettyClientMap.values().forEach(x -> { if (x != null) { - nettyClients.add(x.userName +":"+ x.token); + nettyClients.add(x.userName + ":" + x.token); } }); res.put("nettyClientMap", nettyClients); + res.put("controllerMap", deviceSessionManager.getControllerMap()); res.put("siteMemory", GlobalMemory.getAllSiteMemory()); return AjaxResult.success(res); } diff --git a/maibu-web-middleware/src/main/java/com/maibu/memory/SystemMonitor.java b/maibu-web-middleware/src/main/java/com/maibu/memory/SystemMonitor.java index 438b933..3850fe4 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/memory/SystemMonitor.java +++ b/maibu-web-middleware/src/main/java/com/maibu/memory/SystemMonitor.java @@ -1,11 +1,13 @@ package com.maibu.memory; +import com.maibu.utils.spring.SpringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; import java.lang.management.ManagementFactory; import java.lang.management.OperatingSystemMXBean; import java.lang.management.ThreadMXBean; @@ -19,6 +21,8 @@ import java.util.concurrent.TimeUnit; public class SystemMonitor { private final Logger logger = LoggerFactory.getLogger(SystemMonitor.class); + private ScheduledExecutorService executor = SpringUtils.getBean("scheduledExecutorService"); + // 启动监控任务 @PostConstruct public void startMonitoring() { @@ -37,7 +41,6 @@ public class SystemMonitor { // 内存监控 private void startMemoryMonitor() { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -> { Runtime runtime = Runtime.getRuntime(); long totalMemory = runtime.totalMemory() / (1024 * 1024); @@ -59,7 +62,6 @@ public class SystemMonitor { // CPU 监控 private void startCpuMonitor() { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -> { try { // 使用 OperatingSystemMXBean 获取 CPU 使用率 @@ -84,7 +86,6 @@ public class SystemMonitor { // 线程监控 private void startThreadMonitor() { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -> { ThreadMXBean threadBean = ManagementFactory.getThreadMXBean(); int threadCount = threadBean.getThreadCount(); @@ -105,7 +106,6 @@ public class SystemMonitor { // 连接数监控 private void startConnectionMonitor() { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -> { int webSocketCount = MiddleGlobalMemory.onlineSockets.size(); int nettyClientCount = MiddleGlobalMemory.nettyClientMap.size(); @@ -114,4 +114,5 @@ public class SystemMonitor { webSocketCount, nettyClientCount); }, 1, 5, TimeUnit.MINUTES); } + } \ No newline at end of file diff --git a/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java b/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java index 7dda615..58f9c2d 100644 --- a/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java +++ b/maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java @@ -4,6 +4,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Objects; import java.util.concurrent.TimeUnit; import com.maibu.core.business.dto.WebStatusMessageDTO; @@ -63,10 +64,10 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ctx.writeAndFlush(data); } NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key); - if (nettyClient != null && StringUtils.isEmpty(nettyClient.currentDeviceId)) { + if (nettyClient != null) { if (nettyClient.lastSwitchTime == -1 || System.currentTimeMillis() - nettyClient.lastSwitchTime > nettyClient.Interval) { - if (!StringUtils.isEmpty(nettyClient.requestDeviceId) && !StringUtils.isEmpty(key)) { + if (!StringUtils.isEmpty(nettyClient.requestDeviceId) && !StringUtils.isEmpty(key) && !Objects.equals(nettyClient.requestDeviceId, nettyClient.currentDeviceId)) { switchDevice(nettyClient, key); nettyClient.lastSwitchTime = System.currentTimeMillis(); } @@ -138,7 +139,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { } } else { /* 判断 respond 字段 */ - if (obj.get("request") != null) { + if (obj.get("request") != null && !"null".equals(obj.get("request").toString())) { String deviceId = obj.get("deviceId").toString(); String userName = obj.get("userId").toString(); String platform = obj.get("platform").toString(); @@ -151,7 +152,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { logger.debug("json: {}", JsonUtils.toJsonString(request)); WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(request)); } else { - if (obj.get("respond") != null) { + if (obj.get("respond") != null && !"null".equals(obj.get("respond").toString())) { if ("have_logged_in".equals(obj.get("respond").toString())) { /* 如果回复 have_logged_in 则调用logout */ // System.out.println("检测到重复登录，执行 logout"); @@ -460,7 +461,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { String key = ctx.channel().attr(MiddleConstant.ATT_MASTER_KEY).get(); NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key); // stopHeartbeat(nettyClient); - nettyClient.currentDeviceId = null; + if (nettyClient != null) { + nettyClient.currentDeviceId = null; + } logger.debug("连接断开channelInactive,ChannelId:{} ", ctx.channel().id().asLongText()); WebSocket webSocket = MiddleGlobalMemory.onlineSockets.get(key); if (webSocket != null && webSocket.isOpen()) { 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 db51237..5dc9ce3 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 @@ -10,6 +10,7 @@ import com.maibu.netty.NettyClient; import com.maibu.utils.CommandUtils; import com.maibu.utils.ControlUtils; import com.maibu.utils.json.JsonUtils; +import com.maibu.utils.spring.SpringUtils; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.apache.commons.lang3.StringUtils; @@ -41,7 +42,7 @@ public class WebsocketHandler extends WebSocketServer { // 空闲超时时间（10秒）：超过10秒无消息则判定为断开 private static final long IDLE_TIMEOUT = 10 * 1000; // 定时检测线程池（每隔10秒检测一次空闲连接） - private final ScheduledExecutorService idleCheckExecutor = Executors.newSingleThreadScheduledExecutor(); + private ScheduledExecutorService executor = SpringUtils.getBean("scheduledExecutorService"); public WebsocketHandler(int port) { @@ -81,24 +82,29 @@ public class WebsocketHandler extends WebSocketServer { return; } //todo 先关闭原先的 - WebSocket oldWebsocket = MiddleGlobalMemory.onlineSockets.get(key); - NettyClient oldNettyClient = MiddleGlobalMemory.nettyClientMap.get(key); - if (oldWebsocket != null && oldWebsocket.isOpen()) { - oldWebsocket.close(); - } - if (oldNettyClient != null && oldNettyClient.isConnected()) { - oldNettyClient.disconnect(); - } +// 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客户端 - NettyClient client = new NettyClient(); + if (client == null) { + client = new NettyClient(); + } client.key = key; client.requestDeviceId = deviceId; client.token = token; client.userName = userName; - client.initClient(); - client.initConnect(MiddleConstant.vertxIp, 9001); MiddleGlobalMemory.nettyClientMap.put(key, client); + if (!client.isConnected()) { + client.initClient(); + client.initConnect(MiddleConstant.vertxIp, 9001); + } } else if (MesType.path.equals(type)) { //todo下发路径 预留 pathDistribute(); @@ -126,7 +132,7 @@ public class WebsocketHandler extends WebSocketServer { // else if(MesType.heartbeat.equals(type)){ // clientActiveTimeMap.put(conn,System.currentTimeMillis()); // } - else if(MesType.switchPermission.equals(type)){ + else if (MesType.switchPermission.equals(type)) { DeviceRequestDTO requestDTO = new DeviceRequestDTO(); requestDTO.setRequest(CommandRequestType.switch_control); requestDTO.setToken(token); @@ -136,12 +142,12 @@ public class WebsocketHandler extends WebSocketServer { logger.debug("Netty客户端:{}发送操控请求数据: {}", userName, JsonUtils.toJsonString(requestDTO)); NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key); ByteBuf buf = Unpooled.buffer(); - CommandUtils.buildCommand(buf,requestDTO, CommandConstant.interaction); + CommandUtils.buildCommand(buf, requestDTO, CommandConstant.interaction); if (nettyClient != null && nettyClient.isConnected()) { // if (!nettyClient.isConnected()) nettyClient.connect(); nettyClient.sendData(buf.array()); } - } else if(MesType.switchResult.equals(type)){ + } else if (MesType.switchResult.equals(type)) { DeviceRequestDTO requestDTO = new DeviceRequestDTO(); requestDTO.setToken(token); requestDTO.setPlatform("web"); @@ -154,7 +160,7 @@ public class WebsocketHandler extends WebSocketServer { logger.debug("Netty客户端:{}发送切换操控权请求反馈: {}", userName, JsonUtils.toJsonString(requestDTO)); NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key); ByteBuf buf = Unpooled.buffer(); - CommandUtils.buildCommand(buf,requestDTO, CommandConstant.interaction); + CommandUtils.buildCommand(buf, requestDTO, CommandConstant.interaction); if (nettyClient != null && nettyClient.isConnected()) { // if (!nettyClient.isConnected()) nettyClient.connect(); nettyClient.sendData(buf.array()); @@ -225,7 +231,7 @@ public class WebsocketHandler extends WebSocketServer { // 核心：定时检测空闲连接（每隔10秒执行一次） private void startIdleCheckTask() { - idleCheckExecutor.scheduleAtFixedRate(() -> { + executor.scheduleAtFixedRate(() -> { long now = System.currentTimeMillis(); // 遍历所有连接，检测是否超时 for (Map.Entry entry : clientActiveTimeMap.entrySet()) {