diff --git a/maibu-admin/src/main/resources/application-dev.yml b/maibu-admin/src/main/resources/application-dev.yml index 19aea1d..f0e6540 100644 --- a/maibu-admin/src/main/resources/application-dev.yml +++ b/maibu-admin/src/main/resources/application-dev.yml @@ -3,7 +3,7 @@ spring: datasource: type: com.alibaba.druid.pool.DruidDataSource driver-class-name: com.mysql.cj.jdbc.Driver - url: jdbc:mysql://localhost:3306/maibu?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8 + url: jdbc:mysql://1.95.137.212:3306/maibu?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8 username: satabot password: satabot @@ -31,7 +31,7 @@ spring: mqtt: username: maibu # 账号 password: jsmbzn520 # 密码 - host-url: tcp://localhost:1883 # mqtt连接tcp地址 + host-url: tcp://1.95.137.212:1883 # mqtt连接tcp地址 client-id: maibu-mqtt-client # 客户端Id,不能相同,采用随机数 ${random.value} default-topic: test # 默认主题 timeout: 30 # 超时时间 diff --git a/maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java b/maibu-common/src/main/java/com/maibu/core/business/dto/WebStatusMessageDTO.java similarity index 87% rename from maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java rename to maibu-common/src/main/java/com/maibu/core/business/dto/WebStatusMessageDTO.java index 68525fa..be69899 100644 --- a/maibu-netty-server/src/main/java/com/maibu/dto/WebStatusMessageDTO.java +++ b/maibu-common/src/main/java/com/maibu/core/business/dto/WebStatusMessageDTO.java @@ -1,4 +1,4 @@ -package com.maibu.dto; +package com.maibu.core.business.dto; import lombok.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 17f5474..3e58035 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 @@ -3,6 +3,7 @@ package com.maibu.core.enums; public enum MesType { AUTH, // 登录连接权限web->be deviceInfo, // 设备状态 be->web + locationInfo, path, // 路径下发 web->be have_logged_in, // 消息 device_status_changed, // 上下线状态变化 diff --git a/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java b/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java index 1397845..da31df5 100644 --- a/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java +++ b/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java @@ -87,6 +87,20 @@ public class DeviceSessionManager { System.out.println("Registered device: " + deviceId + " type: " + deviceType); } + /** + * 注册下位机设备 + * + * @param deviceId 设备ID + */ + public void registerSlaveDevice(String deviceId, ConnectorStatus status, String slaveId) { + NettyDevice nettyDevice = deviceMap.computeIfAbsent(deviceId, id -> new NettyDevice(id, ConnectorType.SLAVE)); + nettyDevice.setConnectorStatus(status); // 默认上线即ACTIVE + nettyDevice.setCurrentSlaveId(slaveId); + nettyDevice.setLatestLoginTime(System.currentTimeMillis()); + deviceMap.put(deviceId, nettyDevice); + logger.debug("Registered slave device:{},type:{}", deviceId, ConnectorType.SLAVE); + } + /** * 上位机绑定下位机,建立双向绑定关系 * diff --git a/maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java b/maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java index 49686cc..ab57216 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java @@ -7,12 +7,18 @@ import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; /** * 自定义外部设备 MQTT 订阅管理器 @@ -69,6 +75,22 @@ public class CustomMqttDeviceMonitor { private final ConcurrentHashMap hearBeatMap = new ConcurrentHashMap<>(); + private final ThreadPoolExecutor messageExecutor = new ThreadPoolExecutor( + 4, 8, 30L, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(256), + new ThreadFactory() { + private final AtomicInteger counter = new AtomicInteger(1); + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, "mqtt-message-worker-" + counter.getAndIncrement()); + t.setDaemon(true); + return t; + } + }, + new ThreadPoolExecutor.CallerRunsPolicy() + ); + /** * 巡检间隔(秒) */ @@ -170,7 +192,7 @@ public class CustomMqttDeviceMonitor { } public List getAllHeartBeat() { - return (List) hearBeatMap.values(); + return new ArrayList<>(hearBeatMap.values()); } @@ -202,6 +224,7 @@ public class CustomMqttDeviceMonitor { if (scheduler != null) { scheduler.shutdown(); } + messageExecutor.shutdown(); clientMap.keySet().forEach(this::disconnectDevice); log.info("CustomMqttDeviceMonitor 已停止"); } @@ -248,7 +271,7 @@ public class CustomMqttDeviceMonitor { public void messageArrived(String topic, MqttMessage message) { String payload = new String(message.getPayload()); log.debug("收到消息: deviceKey={}, topic={}, payload={}", deviceKey, topic, payload); - dispatchMessage(deviceKey, topic, payload, config); + messageExecutor.submit(() -> dispatchMessage(deviceKey, topic, payload, config)); } @Override @@ -395,6 +418,7 @@ public class CustomMqttDeviceMonitor { /** * 向指定设备发布消息(指定 QoS) + * 异步投递到线程池执行,避免阻塞 MQTT 回调线程 */ public void publish(String deviceKey, String topic, String payload, int qos) throws MqttException { MqttClient client = clientMap.get(deviceKey); 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 f6cd9a0..8022b73 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java @@ -3,7 +3,11 @@ package com.maibu.mqtt; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.maibu.constant.Constant; +import com.maibu.core.business.device.DeviceStatusDetail; 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; @@ -14,7 +18,10 @@ import com.maibu.utils.StringUtils; import com.maibu.utils.json.JsonUtils; import com.maibu.utils.spring.SpringUtils; import lombok.extern.slf4j.Slf4j; -import org.eclipse.paho.client.mqttv3.MqttException; +import org.springframework.util.CollectionUtils; + +import java.util.ArrayList; +import java.util.List; /** @@ -27,6 +34,10 @@ public class HostMessageHandler implements CustomMqttMessageHandler { private final DeviceSessionManager deviceSessionManager = SpringUtils.getBean(DeviceSessionManager.class); + private final WebsocketMesDispather websocketMesDispather = SpringUtils.getBean(WebsocketMesDispather.class); + + + private final String mqttClientId = "maibu-mqtt-client"; @Override public String getHandlerType() { @@ -105,22 +116,111 @@ public class HostMessageHandler implements CustomMqttMessageHandler { try { log.debug("【位置数据】deviceId={}, payload={}", deviceId, payload); HostLocationRelTimeDTO locationRelTimeDTO = JsonUtils.parseObject(payload, HostLocationRelTimeDTO.class); - - - //todo 发送位置信息 - + if (locationRelTimeDTO != null) { + WebStatusMessageDTO dto = createWebDeviceLocationMessage(locationRelTimeDTO); + List controlMasters = deviceSessionManager + .getAllSlaveControl(deviceId); + if (!CollectionUtils.isEmpty(controlMasters)) { + for (NettyDevice x : controlMasters) { + try { + websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto)); + } catch (Exception e) { + log.error("位置数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(), e.getMessage()); + } + } + } +// GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_LOCATION_TOPIC, deviceId), +// JsonUtils.toJsonString(dto)); + } } catch (Exception e) { log.error("位置消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage()); } } + + public WebStatusMessageDTO createWebDeviceLocationMessage(HostLocationRelTimeDTO locationRelTimeDTO) { + WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO(); + List transferData = new ArrayList<>(); + DeviceStatusDetail lat = new DeviceStatusDetail(); + lat.setName("latitude"); + lat.setValue(String.valueOf(locationRelTimeDTO.getData().getLatitude())); + lat.setUnit(""); + + transferData.add(lat); + + DeviceStatusDetail lng = new DeviceStatusDetail(); + lng.setName("longitude"); + lng.setValue(String.valueOf(locationRelTimeDTO.getData().getLongitude())); + lng.setUnit(""); + transferData.add(lng); + + webStatusMessageDTO.setData(transferData); + webStatusMessageDTO.setType(MesType.locationInfo); + return webStatusMessageDTO; + } + + public WebStatusMessageDTO createWebDeviceRealtimeMessage(HostVehicleRelTimeDTO vehicleRelTimeDTO) { + //todo 状态消息 + + WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO(); + List transferData = new ArrayList<>(); + DeviceStatusDetail voltage = new DeviceStatusDetail(); + voltage.setName("voltage"); + voltage.setValue(String.valueOf(vehicleRelTimeDTO.getData().getBattery_voltage())); + voltage.setUnit(""); + + transferData.add(voltage); + + DeviceStatusDetail speed = new DeviceStatusDetail(); + speed.setName("speed"); + speed.setValue(String.valueOf(vehicleRelTimeDTO.getData().getSpeed())); + speed.setUnit(""); + transferData.add(speed); + + DeviceStatusDetail motion = new DeviceStatusDetail(); + motion.setName("motion"); + motion.setValue(String.valueOf(vehicleRelTimeDTO.getData().getMotion())); + motion.setUnit(""); + transferData.add(motion); + + DeviceStatusDetail percentage = new DeviceStatusDetail(); + percentage.setName("percentage"); + percentage.setValue(String.valueOf(vehicleRelTimeDTO.getData().getBattery_percentage())); + percentage.setUnit(""); + transferData.add(percentage); + + webStatusMessageDTO.setData(transferData); + webStatusMessageDTO.setType(MesType.deviceInfo); + return webStatusMessageDTO; + } + /** * 上位机数据 */ private void handleRealtimePost(String deviceId, String topic, String payload) { log.debug("【实时数据】deviceId={}, payload={}", deviceId, payload); + try { + HostVehicleRelTimeDTO vehicleRelTimeDTO = JsonUtils.parseObject(payload, HostVehicleRelTimeDTO.class); + if (vehicleRelTimeDTO != null) { + WebStatusMessageDTO dto = createWebDeviceRealtimeMessage(vehicleRelTimeDTO); + List controlMasters = deviceSessionManager + .getAllSlaveControl(deviceId); + if (!CollectionUtils.isEmpty(controlMasters)) { + for (NettyDevice x : controlMasters) { + try { + websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto)); + } catch (Exception e) { + log.error("实时数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(), e.getMessage()); + } + } + } +// GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_STATUS_TOPIC, deviceId), +// JsonUtils.toJsonString(dto)); - HostVehicleRelTimeDTO vehicleRelTimeDTO = JsonUtils.parseObject(payload, HostVehicleRelTimeDTO.class); + } + } catch (Exception e) { + log.error("实时消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage()); + } } /** @@ -136,7 +236,7 @@ public class HostMessageHandler implements CustomMqttMessageHandler { * topic: mower/{deviceId}/event/heartbeat/post */ private void handleHeartbeatPost(String deviceId, String topic, String payload) { - log.debug("【心跳】deviceId={}", deviceId); + log.debug("【心跳】payload={}", payload); // 心跳处理 HeartBeatDTO beatDTO = JsonUtils.parseObject(payload, HeartBeatDTO.class); if (beatDTO != null) { @@ -156,7 +256,7 @@ public class HostMessageHandler implements CustomMqttMessageHandler { /** * 处理任务相关上报 */ - private void handleTaskPost(String deviceId, String topic, String payload, String action) throws MqttException { + private void handleTaskPost(String deviceId, String topic, String payload, String action) { log.info("【任务上报】deviceId={}, action={}, payload={}", deviceId, action, payload); 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 cc981ae..d27f192 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java @@ -5,6 +5,8 @@ public class MqttTopic { /** @deprecated 通用设备主题,建议使用 MOWER_* 系列主题 */ public static final String DEVICE_STATUS_TOPIC = "device/%s/realTimeMessage"; + + public static final String DEVICE_LOCATION_TOPIC = "device/%s/locationMessage"; /** @deprecated 通用设备主题,建议使用 MOWER_* 系列主题 */ public static final String DEVICE_TASK_STATUS_TOPIC = "task/%s/status"; /** @deprecated 通用设备主题,建议使用 MOWER_* 系列主题 */ @@ -90,9 +92,9 @@ public class MqttTopic { /** 订阅所有设备的任务状态变更 */ public static final String MOWER_WILDCARD_TASK_STATUS = "mower/+/task/status/post"; - public static final String MOWER_WILDCARD_TASK_ROUTE = " mower/+/task/route/post"; + public static final String MOWER_WILDCARD_TASK_ROUTE = "mower/+/task/route/post"; - public static final String MOWER_HEARTBEAT = " mower/+/event/heartbeat/post"; + public static final String MOWER_HEARTBEAT = "mower/+/event/heartbeat/post"; diff --git a/maibu-framework/src/main/java/com/maibu/config/SecurityConfig.java b/maibu-framework/src/main/java/com/maibu/config/SecurityConfig.java index d305fff..603b5d3 100644 --- a/maibu-framework/src/main/java/com/maibu/config/SecurityConfig.java +++ b/maibu-framework/src/main/java/com/maibu/config/SecurityConfig.java @@ -116,7 +116,7 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter { // 过滤请求 .authorizeRequests() // 对于登录login 注册register 验证码captchaImage 允许匿名访问 - .antMatchers("/api/thermal/**","/iotDA/device/**", "/websocket/allMemory", "/iot/UAV/**", "/iot/api/**", "/iot/device/userDevice", "/external/sms/sendAliSmsCode", "/login", "/register", "/captchaImage", "/iot/tool/register", "/iot/tool/ntp", + .antMatchers("/iot/host/upload-path-csv","/api/thermal/**","/iotDA/device/**", "/websocket/allMemory", "/iot/UAV/**", "/iot/api/**", "/iot/device/userDevice", "/external/sms/sendAliSmsCode", "/login", "/register", "/captchaImage", "/iot/tool/register", "/iot/tool/ntp", "/iot/tool/mqtt/auth", "/iot/workRecord/**", "/iot/tool/mqtt/authv5", "/iot/tool/mqtt/webhook", "/iot/tool/mqtt/webhookv5", "/auth/**/**", "/wechat/mobileLogin", "/wechat/miniLogin", "/wechat/wxBind/callback").permitAll() .antMatchers("/zlmhook/**").permitAll() diff --git a/maibu-netty-server/src/main/java/com/maibu/netty/handler/CommandForwardHandler.java b/maibu-netty-server/src/main/java/com/maibu/netty/handler/CommandForwardHandler.java index c6c6ca1..e4eeb00 100644 --- a/maibu-netty-server/src/main/java/com/maibu/netty/handler/CommandForwardHandler.java +++ b/maibu-netty-server/src/main/java/com/maibu/netty/handler/CommandForwardHandler.java @@ -4,10 +4,13 @@ import java.util.List; import java.util.Set; import com.maibu.memory.DeviceSessionManager; +import com.maibu.memory.GlobalMemory; +import com.maibu.mqtt.MqttTopic; import org.apache.commons.codec.binary.Hex; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; @@ -32,6 +35,9 @@ public class CommandForwardHandler extends SimpleChannelInboundHandler { @Autowired private DeviceSessionManager deviceSessionManager; + @Value("${mqtt.client-id}") + private String mqttClientId; + @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { Channel currentChannel = ctx.channel(); @@ -64,12 +70,6 @@ public class CommandForwardHandler extends SimpleChannelInboundHandler { } if (ConnectorType.MASTER.equals(senderNettyDevice.getDeviceType())) { - - - - //TODO 有控制权才能发 - - Set connectedDevices = senderNettyDevice.getConnectedConnectors(); if (connectedDevices == null || connectedDevices.isEmpty()) { logger.debug("发送拒绝master:{},该设备没有关联的设备", senderDeviceId); @@ -84,11 +84,14 @@ public class CommandForwardHandler extends SimpleChannelInboundHandler { logger.debug("发送拒绝master:{},消息到达的slave设备:{},处于禁止接收状态", senderDeviceId, target.getConnectorId()); continue; } - Channel targetChannel = target.getChannel(); - if (targetChannel != null && targetChannel.isActive()) { - targetChannel.writeAndFlush(msg); - logger.debug("从master:{},向slave:{}, 转发消息:{}", senderDeviceId, target.getConnectorId(), msg); - } + // todo 向上位机发送控制消息 + + GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.MOWER_CONTROL_SET, target.getConnectorId()), msg); +// Channel targetChannel = target.getChannel(); +// if (targetChannel != null && targetChannel.isActive()) { +// targetChannel.writeAndFlush(msg); +// logger.debug("从master:{},向slave:{}, 转发消息:{}", senderDeviceId, target.getConnectorId(), msg); +// } } } else if (ConnectorType.SLAVE.equals(senderNettyDevice.getDeviceType())) { List masters = deviceSessionManager.getMastersBySlaveId(senderDeviceId); diff --git a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java index 69b408d..75a00bc 100644 --- a/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java +++ b/maibu-netty-server/src/main/java/com/maibu/netty/handler/DataToDataBaseHandler.java @@ -13,10 +13,8 @@ import com.maibu.constant.Constant; import com.maibu.core.business.DeviceRunParam; import com.maibu.mapper.DeviceRunParamMapper; import com.maibu.memory.DeviceSessionManager; -import com.maibu.memory.SiteMemory; import org.apache.commons.codec.binary.Hex; import org.apache.commons.lang3.StringUtils; -import org.eclipse.paho.client.mqttv3.MqttException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -34,7 +32,7 @@ import com.maibu.core.business.device.NettyDevice; import com.maibu.core.business.inter.WebsocketMesDispather; import com.maibu.core.enums.MesType; import com.maibu.core.redis.RedisCache; -import com.maibu.dto.WebStatusMessageDTO; +import com.maibu.core.business.dto.WebStatusMessageDTO; import com.maibu.influxdb.MowerRealTimeData; import com.maibu.influxdb.util.InfluxSqlBuilder; import com.maibu.memory.GlobalMemory; @@ -101,15 +99,8 @@ public class DataToDataBaseHandler extends SimpleChannelInboundHandler { String deviceStatus = new String(bytes, 2, bytes.length - 4, StandardCharsets.UTF_8).trim(); String[] split = deviceStatus.split(","); WebStatusMessageDTO dto = createWebDeviceStatusMessage(split); - try { - GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_STATUS_TOPIC, deviceID), + GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_STATUS_TOPIC, deviceID), JsonUtils.toJsonString(dto)); -// GlobalMemory.mqttClientUtil.publish( -// String.format(MqttTopic.DEVICE_STATUS_TOPIC, deviceID), -// JsonUtils.toJsonString(dto)); - } catch (MqttException e) { - logger.error("发送实时状态数据错误: {}", e.getMessage()); - } List controlMasters = deviceSessionManager .getAllSlaveControl(deviceID); if (!CollectionUtils.isEmpty(controlMasters)) { 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 a6cf0a9..2d00286 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 @@ -194,51 +194,6 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { handleMasterConnect(ctx, sessionId, null, ConnectorStatus.ACTIVE); } } - } else { - handleSlaveConnect(ctx, data); - logger.info("{}: 下位机连接", data); - Device device = deviceService.selectDeviceBySerialNumber(data); - if (device == null) { - device = new Device(); - device.setStatus(3); - device.setOnlineStatus(DeviceStatus.online.getType()); - device.setSerialNumber(data); - device.setDeviceName(data); - device.setProductId(-1L); - device.setTenantId(-1L); - device.setFirmwareVersion(BigDecimal.valueOf(1.0)); - device.setProductName("割草机产品MC700"); - device.setCreateTime(LocalDateTime.now()); - device.setUpdateTime(LocalDateTime.now()); - deviceService.insertDeviceBySelf(device); - } else { - device.setStatus(3); - device.setOnlineStatus(DeviceStatus.online.getType()); - device.setUpdateTime(LocalDateTime.now()); - deviceService.updateDeviceBySelf(device); - } - NettyDevice slaveDevice = sessionManager.getDevice(data); - if (slaveDevice != null) { - slaveDevice.setOnlineStatus(1); - slaveDevice.setDevice(device); - } - // 推送json 找到设备绑定的主机进行推送 - List controlMasters = sessionManager.getAllSlaveControl(device.getSerialNumber()); - if (!CollectionUtils.isEmpty(controlMasters)) { - controlMasters.forEach(x -> { - DeviceStatusChangeDTO changeDTO = new DeviceStatusChangeDTO(); - changeDTO.setDeviceId(data); - changeDTO.setStatus(DeviceStatus.online.getCode()); - changeDTO.setEvent(RespondCode.device_status_changed); - ByteBuf buf = ctx.alloc().buffer(); - CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction); - String content = buf.toString(CharsetUtil.UTF_8); - logger.info("推送下位机登录状态消息:{}", content); - if (x.getChannel() != null && x.getChannel().isActive()) { - x.getChannel().writeAndFlush(buf); - } - }); - } } } } else if (cmdType == CommandConstant.interaction) { @@ -285,7 +240,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { } } - public void pushFinishData(NettyDevice device, String deviceId) { + public void pushFinishData(NettyDevice device, String deviceId) throws MqttException { Long taskId = device.finishTask(); // todo 推送信息给前端 DeviceTaskStatusMessageDTO changeDTO = new DeviceTaskStatusMessageDTO(); @@ -294,16 +249,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { changeDTO.setType(MesType.device_task_change); changeDTO.setTaskId(taskId); - try { - - GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId), + GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId), JsonUtils.toJsonString(changeDTO)); - -// GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId), -// JsonUtils.toJsonString(changeDTO)); - } catch (MqttException e) { - logger.error("任务完成状态推送失败 task:{},device:{}error:{}", taskId, deviceId, e.getMessage()); - } List controlMasters = sessionManager .getAllSlaveControl(deviceId); if (!CollectionUtils.isEmpty(controlMasters)) { @@ -317,7 +264,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { } } - public void pushArriveData(NettyDevice device, String deviceId) { + public void pushArriveData(NettyDevice device, String deviceId) throws MqttException { Long taskId = device.getTask() == null ? null : device.getTask().getId(); // todo 推送信息给前端 DeviceTaskStatusMessageDTO reportDTO = new DeviceTaskStatusMessageDTO(); @@ -325,15 +272,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler { reportDTO.setTaskId(taskId); reportDTO.setEntity(device.getCurrentPoint()); reportDTO.setType(MesType.device_task_arrive_point); - try { - - GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_ARRIVE_TOPIC, taskId), + GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_ARRIVE_TOPIC, taskId), JsonUtils.toJsonString(reportDTO)); -// GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_TASK_ARRIVE_TOPIC, taskId), -// JsonUtils.toJsonString(reportDTO)); - } catch (MqttException e) { - logger.error("任务点位到达推送失败 task:{},device:{}error:{}", taskId, deviceId, e.getMessage()); - } List controlMasters = sessionManager .getAllSlaveControl(deviceId); 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 3b29c7b..2c5fdce 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 @@ -2,6 +2,7 @@ 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; @@ -14,12 +15,19 @@ 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.eclipse.paho.client.mqttv3.MqttException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.core.Constants; @@ -37,9 +45,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.ErrorSource; -import com.maibu.core.enums.RespondCode; import com.maibu.core.redis.RedisCache; import com.maibu.dto.DeviceErrorPushDTO; import com.maibu.mapper.ErrorIdentificationStandardMapper; @@ -74,6 +79,9 @@ public class DeviceThreadService { @Autowired private AlertPushService alertPushService; + @Autowired + private IDeviceService deviceService; + @Value("${mqtt.client-id}") private String mqttClientId; @@ -201,7 +209,7 @@ public class DeviceThreadService { log.info("推送错误信息 device:{},content:{}", deviceId, dto); }); } - } catch (MqttException e) { + } catch (Exception e) { log.error("推送错误信息失败 device:{}error:{}", deviceId, e.getMessage()); } } @@ -270,19 +278,58 @@ public class DeviceThreadService { List heartBeatDTOS = GlobalMemory.customMqttDeviceMonitor.getAllHeartBeat(); if (!CollectionUtils.isEmpty(heartBeatDTOS)) { heartBeatDTOS.forEach(x -> { + String deviceId = x.getSn(); Long lastActiveTime = x.getSysTimestamp(); - NettyDevice device = sessionManager.getDevice(x.getSn()); + NettyDevice nettyDevice = sessionManager.getDevice(x.getSn()); if (nowTime - lastActiveTime >= Constant.hostOfflineInterval) { //超时了判断离线 - if (device != null) { - device.status(0); + if (nettyDevice != null) { + nettyDevice.status(0); //todo 推送设备离线 - } } else { - - - + if (nettyDevice == null) { + sessionManager.registerSlaveDevice(deviceId, ConnectorStatus.ACTIVE, null); + Device device = deviceService.selectDeviceBySerialNumber(deviceId); + if (device == null) { + device = new Device(); + device.setStatus(3); + device.setOnlineStatus(DeviceStatus.online.getType()); + device.setSerialNumber(deviceId); + device.setDeviceName(deviceId); + device.setProductId(-1L); + device.setTenantId(-1L); + device.setFirmwareVersion(BigDecimal.valueOf(1.0)); + device.setProductName("割草机产品MC700"); + device.setCreateTime(LocalDateTime.now()); + device.setUpdateTime(LocalDateTime.now()); + deviceService.insertDeviceBySelf(device); + } else { + device.setStatus(3); + device.setOnlineStatus(DeviceStatus.online.getType()); + device.setUpdateTime(LocalDateTime.now()); + deviceService.updateDeviceBySelf(device); + } + // 推送json 找到设备绑定的主机进行推送 +// List controlMasters = sessionManager.getAllSlaveControl(device.getSerialNumber()); +// if (!CollectionUtils.isEmpty(controlMasters)) { +// controlMasters.forEach(x -> { +// DeviceStatusChangeDTO changeDTO = new DeviceStatusChangeDTO(); +// changeDTO.setDeviceId(data); +// changeDTO.setStatus(DeviceStatus.online.getCode()); +// changeDTO.setEvent(RespondCode.device_status_changed); +// ByteBuf buf = ctx.alloc().buffer(); +// CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction); +// String content = buf.toString(CharsetUtil.UTF_8); +// logger.info("推送下位机登录状态消息:{}", content); +// if (x.getChannel() != null && x.getChannel().isActive()) { +// x.getChannel().writeAndFlush(buf); +// } +// }); + } else { + //todo 推送设备上线 + nettyDevice.status(1); + } } }); } @@ -290,8 +337,6 @@ public class DeviceThreadService { log.error("执行监测心跳线程错误:{}", e.getMessage()); } }, 0, 500, TimeUnit.MILLISECONDS); - - } } 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 60e62fd..7dda615 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 @@ -6,6 +6,7 @@ import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; +import com.maibu.core.business.dto.WebStatusMessageDTO; import org.apache.commons.lang3.StringUtils; import org.java_websocket.WebSocket; import org.slf4j.Logger; @@ -23,7 +24,6 @@ import com.maibu.dto.ResultDTO; import com.maibu.dto.WebAuthResponseDTO; import com.maibu.dto.WebDeviceTaskStatusMessageDTO; import com.maibu.dto.WebOlineStatusMessageDTO; -import com.maibu.dto.WebStatusMessageDTO; import com.maibu.dto.WebSwitchControlResponseDTO; import com.maibu.memory.MiddleGlobalMemory; import com.maibu.netty.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 e31308f..7154bb1 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 @@ -13,7 +13,7 @@ public class WebsocketMesHandler implements DeviceWebSocketInterace { public void sendWebsoketMes(String key, String data) { MiddleGlobalMemory.onlineSockets.keySet().forEach(x -> { // key: test:web:token - if (x.startsWith(key)) { + if (x.contains(key)) { WebSocket webSocket = MiddleGlobalMemory.onlineSockets.get(x); WebsocketHandler.sendMessageToClient(webSocket, data); }