IOT/MiddlePlatform
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

# 上位机心跳 以及定位信息发送部分

Browse Source
This commit is contained in:
rqian
2026-08-19 16:57:55 +08:00
parent 2f94639f0a
commit 16ec98e36a
14 changed files with 238 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
maibu-common/src/main/java/com/maibu/core/business/dto/WebStatusMessageDTO.java Normal file
View File

@@ -0,0 +1,14 @@
package com.maibu.core.business.dto;
import lombok.Data;
import java.util.List;
import com.maibu.core.business.device.DeviceStatusDetail;
import com.maibu.core.enums.MesType;
@Data
public class WebStatusMessageDTO {
private List<DeviceStatusDetail> data;
private MesType type;
}

1
maibu-common/src/main/java/com/maibu/core/enums/MesType.java
View File

@@ -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, // 上下线状态变化

14
maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java
View File

@@ -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);
}
/**
* 上位机绑定下位机，建立双向绑定关系
*

28
maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java
View File

@@ -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<String, HeartBeatDTO> 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<HeartBeatDTO> getAllHeartBeat() {
return (List<HeartBeatDTO>) 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);

116
maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java
View File

@@ -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<NettyDevice> 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<DeviceStatusDetail> 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<DeviceStatusDetail> 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<NettyDevice> 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);

6
maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java
View File

@@ -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";