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

# update 权限相关

Browse Source
This commit is contained in:
rqian
2026-08-25 09:19:12 +08:00
parent 27f4796c55
commit 33df2ff0f7
20 changed files with 222 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
maibu-common/src/main/java/com/maibu/core/business/DevicePlanTask.java
View File

@@ -40,6 +40,6 @@ public class DevicePlanTask extends OrgBaseDO {
private LocalDateTime canExecuteTime; // 这个时间之前都不能执行
@TableField(exist = false)
private boolean taskAccepted = false; //任务被上位机接收
private volatile boolean taskAccepted = false; //任务被上位机接收
}

46
maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java
View File

@@ -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());
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);

8
maibu-common/src/main/java/com/maibu/core/domain/model/LoginUser.java
View File

@@ -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<String> permissions) {
public LoginUser(Long userId, Long deptId, Long orgId, Long siteId, Long roleId, String language, SysUser user, Set<String> permissions) {
this.userId = userId;
this.deptId = deptId;
this.orgId = orgId;

19
maibu-common/src/main/java/com/maibu/core/host/HostNavigationReplyDTO.java Normal file
View File

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

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

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

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

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

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

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

6
maibu-external/src/main/java/com/maibu/uav/service/UAVService.java
View File

@@ -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<String, Long> 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;
}
}

5
maibu-framework/src/main/java/com/maibu/web/service/TokenService.java
View File

@@ -170,8 +170,9 @@ public class TokenService {
Map<String, Object> 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) {

9
maibu-netty-server/src/main/java/com/maibu/controller/TransferDeviceController.java
View File

@@ -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();

6
maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java
View File

@@ -192,6 +192,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
if (masterDevice == null) {
// sessionManager.getControl(sessionId);
handleMasterConnect(ctx, sessionId, null, ConnectorStatus.ACTIVE);
} else {
ctx.channel().attr(Constant.ATT_DEVICE_ID).set(sessionId);
}
}
}
@@ -422,11 +424,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
private void handlePermissionRequest(DeviceRequestDTO requestDTO, byte[] bytes, ChannelHandlerContext ctx) {
logger.info("转发控制请求 :{}", JsonUtils.toJsonString(requestDTO));
String slaveId = requestDTO.getDeviceId();
List<NettyDevice> 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)) {
// 如果当前设备有控制

27
maibu-netty-server/src/main/java/com/maibu/netty/handler/HeartbeatHandler.java
View File

@@ -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) {

9
maibu-netty-server/src/main/java/com/maibu/service/DevicePlanTaskMonitorService.java
View File

@@ -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<String, List<DevicePlan>> 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<SiteMemory> list = GlobalMemory.getAllSiteMemory();
@@ -250,7 +249,6 @@ public class DevicePlanTaskMonitorService {
public void doExecute() {
//prepare 进执行execute的逻辑
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
executor.scheduleWithFixedDelay(() -> {
try {
List<SiteMemory> list = GlobalMemory.getAllSiteMemory();
@@ -270,5 +268,4 @@ public class DevicePlanTaskMonitorService {
}
}, 0, 1000, TimeUnit.MILLISECONDS);
}
}

72
maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java
View File

@@ -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();

6
maibu-netty-server/src/main/java/com/maibu/service/TransferDeviceService.java
View File

@@ -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)) {

17
maibu-service/maibu-system-service/src/main/java/com/maibu/service/impl/IotDeviceCommonService.java
View File

@@ -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<Long, List<DeviceModelPropertyMapping>> mappingCache = new ConcurrentHashMap<>();
public void commonDeviceMonitor() {
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
executor.scheduleWithFixedDelay(() -> {
try {
CustomMqttDeviceMonitor monitor = GlobalMemory.customMqttDeviceMonitor;

3
maibu-web-middleware/src/main/java/com/maibu/controller/WebMiddlewareController.java
View File

@@ -62,10 +62,11 @@ public class WebMiddlewareController extends BaseController {
List<String> 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);
}

9
maibu-web-middleware/src/main/java/com/maibu/memory/SystemMonitor.java
View File

@@ -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);
}
}

11
maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java
View File

@@ -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);
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()) {

38
maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketHandler.java
View File

@@ -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;
MiddleGlobalMemory.nettyClientMap.put(key, client);
if (!client.isConnected()) {
client.initClient();
client.initConnect(MiddleConstant.vertxIp, 9001);
MiddleGlobalMemory.nettyClientMap.put(key, client);
}
} 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<WebSocket, Long> entry : clientActiveTimeMap.entrySet()) {