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

#update mqtt 上位机

Browse Source
This commit is contained in:
rqian
2026-08-07 08:45:43 +08:00
parent c3398c6c07
commit 3ba32b19ce
12 changed files with 403 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
maibu-common/src/main/java/com/maibu/constant/Constant.java
View File

@@ -24,4 +24,6 @@ public class Constant {
public static final String WheelGainCoefficients = "WheelGainCoefficients";
public static final String mqttClientPrefix = "common-device-client-";
public static final String mqttHandleKey = "hostHandle";
}

40
maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java
View File

@@ -1,17 +1,15 @@
package com.maibu.memory;
import com.maibu.constant.Constant;
import com.maibu.core.business.Device;
import com.maibu.core.business.IoTCommonDevice;
import com.maibu.influxdb.util.InfluxDBUtil;
import com.maibu.mqtt.MqttClientUtil;
import com.maibu.mqtt.*;
import lombok.Data;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
@Component
@@ -28,8 +26,9 @@ public class GlobalMemory {
public static InfluxDBUtil influxDBUtil;
public static MqttClientUtil mqttClientUtil;
// public static MqttClientUtil mqttClientUtil;
public static CustomMqttDeviceMonitor customMqttDeviceMonitor = null; //这里面包含了 连我们自己的mqtt 和外部的一些设备，自己的用clientId 是固定的
public static SiteMemory getSiteMemory(Long orgId, Long siteId) {
@@ -52,7 +51,7 @@ public class GlobalMemory {
/**
* 新增场站内存
*
*
* @param orgId
* @param siteId
* @param siteMemory
@@ -72,7 +71,7 @@ public class GlobalMemory {
/**
* 删除
*
*
* @param orgId
* @param siteId
*/
@@ -85,4 +84,29 @@ public class GlobalMemory {
}
}
public static void initMqttDeviceMonitor(String mqttHostUrl, String mqttClientId, String mqttUsername, String mqttPassword) {
if (customMqttDeviceMonitor == null) {
// 1. 创建 MQTT 监控器和消息处理器
customMqttDeviceMonitor = new CustomMqttDeviceMonitor();
// customMqttDeviceMonitor.registerHandler(new GenericPropertyMessageHandler());
customMqttDeviceMonitor.registerHandler(new HostMessageHandler()); //专门处理上位机的
// customMqttDeviceMonitor.registerHandler(new ActionMessageHandler());
//单独注册 对接上位机的
// 注册设备到 MQTT 监控器
CustomMqttDeviceConfig deviceConfig = new CustomMqttDeviceConfig();
deviceConfig.setDeviceKey(mqttClientId);
deviceConfig.setBrokerUrl(mqttHostUrl);
deviceConfig.setClientId(mqttClientId);
deviceConfig.setUsername(mqttUsername);
deviceConfig.setPassword(mqttPassword);
//TODO 添加所有需要订阅的
deviceConfig.setSubscribeTopics(Arrays.asList(MqttTopic.MOWER_WILDCARD_LOCATION));
deviceConfig.setHandlerType(Constant.mqttHandleKey); //唯一的 用于处理上位机mqtt的
customMqttDeviceMonitor.registerDevice(deviceConfig);
}
}
}

2
maibu-common/src/main/java/com/maibu/memory/SiteMemory.java
View File

@@ -30,8 +30,6 @@ public class SiteMemory {
public SysSite sysSite;
public CustomMqttDeviceMonitor CustomMqttDeviceMonitor;
public static final ConcurrentHashMap<String, MqttClient> mqttClientMap = new ConcurrentHashMap<>();
public static final ConcurrentHashMap<Long, OrderModel> orderModelMap = new ConcurrentHashMap<>();

283
maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java Normal file
View File

@@ -0,0 +1,283 @@
package com.maibu.mqtt;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.maibu.constant.Constant;
import lombok.extern.slf4j.Slf4j;
/**
* mqtt相关的 上位机处理器
*/
@Slf4j
public class HostMessageHandler implements CustomMqttMessageHandler {
private final ObjectMapper objectMapper = new ObjectMapper();
@Override
public String getHandlerType() {
return Constant.mqttHandleKey;
}
@Override
public boolean supports(String handlerType) {
return Constant.mqttHandleKey.equals(handlerType);
}
/**
* 处理 MQTT 消息
*
* <p>通配符订阅场景下，deviceKey 为配置中的固定标识（如 "host-monitor"），
* 真正的设备 ID 需从 topic 中解析，通过 {@link #extractDeviceId(String)} 获取。</p>
*/
@Override
public void handleMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config) {
try {
String deviceId = extractDeviceId(topic);
if (deviceId == null) {
log.warn("无法从topic解析deviceId: {}", topic);
return;
}
String action = extractAction(topic);
if (action == null) {
log.debug("未识别的topic: {}", topic);
return;
}
switch (action) {
case "location_post":
handleLocationPost(deviceId, topic, payload);
break;
case "realtime_post":
handleRealtimePost(deviceId, topic, payload);
break;
case "status_post":
handleStatusPost(deviceId, topic, payload);
break;
case "heartbeat_post":
handleHeartbeatPost(deviceId, topic, payload);
break;
case "error_post":
handleErrorPost(deviceId, topic, payload);
break;
case "route_post":
case "point_post":
handleTaskPost(deviceId, topic, payload, action);
break;
case "task_status_post":
case "task_finish_post":
handleTaskPost(deviceId, topic, payload, action);
break;
case "config_reply":
handleConfigReply(deviceId, topic, payload);
break;
case "control_reply":
handleControlReply(deviceId, topic, payload);
break;
default:
log.debug("割草机消息: deviceId={}, topic={}, action={}", deviceId, topic, action);
break;
}
} catch (Exception e) {
log.error("消息处理异常: topic={}, error={}", topic, e.getMessage());
}
}
/**
* 处理 GPS 位置数据上报
* topic: mower/{deviceId}/property/location/post
*/
private void handleLocationPost(String deviceId, String topic, String payload) {
try {
JsonNode jsonNode = objectMapper.readTree(payload);
double latitude = getDouble(jsonNode, "latitude", "lat");
double longitude = getDouble(jsonNode, "longitude", "lng", "lon");
double speed = getDouble(jsonNode, "speed");
double heading = getDouble(jsonNode, "heading", "direction");
double altitude = getDouble(jsonNode, "altitude", "alt");
int satellites = getInt(jsonNode, "satellites", "sats");
String fixQuality = getString(jsonNode, "fixQuality", "fix_quality");
Double batteryLevel = getDoubleObj(jsonNode, "batteryLevel", "battery");
log.info("【位置数据】deviceId={}, lat={}, lng={}, speed={}, heading={}, sats={}",
deviceId, latitude, longitude, speed, heading, satellites);
onLocationUpdate(deviceId, latitude, longitude, speed, heading, altitude, satellites, fixQuality, batteryLevel);
} catch (Exception e) {
log.error("位置消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage());
}
}
/**
* 处理实时运行数据
* topic: mower/{deviceId}/property/realtime/post
*/
private void handleRealtimePost(String deviceId, String topic, String payload) {
log.debug("【实时数据】deviceId={}, payload={}", deviceId, payload);
}
/**
* 处理状态变更事件
* topic: mower/{deviceId}/event/status/post
*/
private void handleStatusPost(String deviceId, String topic, String payload) {
log.info("【状态变更】deviceId={}, payload={}", deviceId, payload);
}
/**
* 处理心跳
* topic: mower/{deviceId}/event/heartbeat/post
*/
private void handleHeartbeatPost(String deviceId, String topic, String payload) {
log.debug("【心跳】deviceId={}", deviceId);
}
/**
* 处理错误告警
* topic: mower/{deviceId}/event/error/post
*/
private void handleErrorPost(String deviceId, String topic, String payload) {
log.warn("【错误告警】deviceId={}, payload={}", deviceId, payload);
}
/**
* 处理任务相关上报
*/
private void handleTaskPost(String deviceId, String topic, String payload, String action) {
log.info("【任务上报】deviceId={}, action={}, payload={}", deviceId, action, payload);
}
/**
* 处理配置响应
*/
private void handleConfigReply(String deviceId, String topic, String payload) {
log.debug("【配置响应】deviceId={}, payload={}", deviceId, payload);
}
/**
* 处理控制指令响应
*/
private void handleControlReply(String deviceId, String topic, String payload) {
log.debug("【控制响应】deviceId={}, payload={}", deviceId, payload);
}
/**
* 位置数据回调，子类可覆盖实现业务逻辑
*/
protected void onLocationUpdate(String deviceId, double latitude, double longitude,
double speed, double heading, double altitude,
int satellites, String fixQuality, Double batteryLevel) {
}
/**
* 从 Topic 中解析 deviceId
* Topic 格式：mower/{deviceId}/...
*/
private String extractDeviceId(String topic) {
if (topic == null || topic.isEmpty()) {
return null;
}
String[] parts = topic.split("/");
if (parts.length < 2 || !"mower".equals(parts[0])) {
return null;
}
return parts[1];
}
/**
* 从 Topic 中提取动作标识
* 将 mower/{deviceId}/xxx/yyy/post 转为 xxx_yyy_post 形式
*/
private String extractAction(String topic) {
String[] parts = topic.split("/");
if (parts.length < 3 || !"mower".equals(parts[0])) {
return null;
}
StringBuilder path = new StringBuilder();
for (int i = 2; i < parts.length; i++) {
if (i > 2) path.append("/");
path.append(parts[i]);
}
String remaining = path.toString();
if (remaining.equals("property/location/post")) return "location_post";
if (remaining.equals("property/realtime/post")) return "realtime_post";
if (remaining.equals("event/status/post")) return "status_post";
if (remaining.equals("event/heartbeat/post")) return "heartbeat_post";
if (remaining.equals("event/error/post")) return "error_post";
if (remaining.equals("task/route/post")) return "route_post";
if (remaining.equals("task/point/post")) return "point_post";
if (remaining.equals("task/status/post")) return "task_status_post";
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";
return null;
}
private double getDouble(JsonNode node, String... names) {
for (String name : names) {
JsonNode fn = node.get(name);
if (fn != null && !fn.isNull()) {
if (fn.isNumber()) return fn.asDouble();
try {
return Double.parseDouble(fn.asText());
} catch (Exception ignored) {
}
}
}
return -1.0;
}
private Double getDoubleObj(JsonNode node, String... names) {
for (String name : names) {
JsonNode fn = node.get(name);
if (fn != null && !fn.isNull()) {
if (fn.isNumber()) return fn.asDouble();
try {
return Double.parseDouble(fn.asText());
} catch (Exception ignored) {
}
}
}
return null;
}
private int getInt(JsonNode node, String... names) {
for (String name : names) {
JsonNode fn = node.get(name);
if (fn != null && !fn.isNull()) {
if (fn.isNumber()) return fn.asInt();
try {
return Integer.parseInt(fn.asText());
} catch (Exception ignored) {
}
}
}
return -1;
}
private String getString(JsonNode node, String... names) {
for (String name : names) {
JsonNode fn = node.get(name);
if (fn != null && !fn.isNull()) {
return fn.asText();
}
}
return null;
}
@Override
public void onConnectionLost(String deviceKey, Throwable cause) {
log.warn("主机监控连接断开: deviceKey={}, reason={}", deviceKey, cause.getMessage());
}
@Override
public void onSubscribed(String deviceKey, String topic) {
log.info("主机监控订阅主题: deviceKey={}, topic={}", deviceKey, topic);
}
}

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

@@ -18,12 +18,12 @@ public class MqttTopic {
// ==================== 设备连接与状态 ====================
/** 设备上线/下线事件上报（设备->平台），对应 TCP 0x03 connect */
public static final String MOWER_STATUS_POST = "mower/%s/event/status/post";
/** 状态变更确认回复（平台->设备） */
// public static final String MOWER_STATUS_REPLY = "mower/%s/event/status/reply";
// ==================== 实时运行数据 ====================
/** 实时运行数据上报：电压/速度/温度/位置等（设备->平台），对应 TCP 0x02 status */
public static final String MOWER_REALTIME_POST = "mower/%s/property/realtime/post";
/** GPS 位置数据上报：经纬度、速度、航向等（设备->平台） */
public static final String MOWER_LOCATION_POST = "mower/%s/property/location/post";
// ==================== 心跳保活 ====================
/** 设备心跳上报（设备->平台），对应 TCP 0xFF heartbeat */
@@ -43,22 +43,12 @@ public class MqttTopic {
/** 远程控制指令下发：前进/后退/转向/割草等（平台->设备），对应 TCP 0x00 remoteControl */
public static final String MOWER_CONTROL_SET = "mower/%s/action/control/set";
// /** 控制指令响应（设备->平台） */
// public static final String MOWER_CONTROL_REPLY = "mower/%s/action/control/reply";
// ==================== 任务管理 ====================
/** 路径规划下发（平台->设备），对应 TCP 0x01 path */
public static final String MOWER_TASK_ROUTE_SET = "mower/%s/task/route/set";
public static final String MOWER_TASK_ROUTE_POST = "mower/%s/task/route/post";
/** 启动任务执行（平台->设备），对应 TCP 0x12 interaction */
public static final String MOWER_TASK_STATUS_SET = "mower/%s/task/status/set";
// /** 暂停任务（平台->设备） */
// public static final String MOWER_TASK_PAUSE_SET = "mower/%s/task/pause/set";
// /** 继续任务（平台->设备） */
// public static final String MOWER_TASK_RESUME_SET = "mower/%s/task/resume/set";
// /** 取消任务（平台->设备） */
// public static final String MOWER_TASK_CANCEL_SET = "mower/%s/task/cancel/set";
// /** 到达路径点上报（设备->平台） */
public static final String MOWER_TASK_POINT_POST = "mower/%s/task/point/post";
/** 任务状态变更上报：执行中/暂停/完成等（设备->平台） */
public static final String MOWER_TASK_STATUS_POST = "mower/%s/task/status/post";
@@ -90,6 +80,8 @@ public class MqttTopic {
public static final String MOWER_WILDCARD_STATUS = "mower/+/event/status/post";
/** 订阅所有设备的实时运行数据 */
public static final String MOWER_WILDCARD_REALTIME = "mower/+/property/realtime/post";
/** 上位机的额外消息 */
public static final String MOWER_WILDCARD_LOCATION = "mower/+/property/location/post";
/** 订阅所有设备的错误告警 */
public static final String MOWER_WILDCARD_ERROR = "mower/+/event/error/post";
/** 订阅所有设备的任务状态变更 */