Sumuxi/fastbee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：最近修改

Browse Source
This commit is contained in:
George 2025-12-10 21:32:30 +08:00
parent a4583c26a6
commit 017c66134d
9 changed files with 378 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
fastbee-admin/src/main/resources/logback.xml
View File

@ -3,7 +3,7 @@
<!-- 日志存放路径 -->
<property name="log.path" scope="context" value="./logs" />
<!-- 日志输出格式 -->
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n" />
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %class - [%method,%line] - %msg%n" />
<!-- 控制台输出 -->
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">

4
fastbee-common/src/main/java/com/fastbee/common/enums/DeviceLogTypeEnum.java
View File

@ -24,7 +24,9 @@ public enum DeviceLogTypeEnum {
DEVICE_UPDATE(4, "设备升级"),
DEVICE_ONLINE(5, "设备上线"),
DEVICE_OFFLINE(6, "设备离线"),
SCENE_VARIABLE_REPORT(7, "场景录入、运算变量上报下发");
SCENE_VARIABLE_REPORT(7, "场景录入、运算变量上报下发"),
DATA_PARSE_ERROR(8, "数据解析错误"),
COMMAND_SEND_ERROR(9, "指令发送错误");;
private final Integer type;

50
fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java
View File

@ -321,6 +321,56 @@ public class ByteUtils {
}
/**
* 将字符串转换为字节数组，强制按十六进制处理
* 清理字符串后直接转换为十六进制字节数组
*
* @param command 命令字符串
* @return 字节数组，转换失败返回空数组
*/
public static byte[] convertToHexByteArray(String command) {
if (command == null || command.isEmpty()) {
return new byte[0];
}
try {
// 清理字符串：去除空格、去除0x前缀、补齐偶数长度
String cleanedCommand = command.trim()
.replaceAll("\\s+", "") // 去除所有空格
.replaceAll("(?i)^0x", ""); // 去除0x或0X前缀
// 如果长度为奇数，在前面补0
if (cleanedCommand.length() % 2 != 0) {
cleanedCommand = "0" + cleanedCommand;
}
// 使用现有的hexToByte方法
byte[] result = hexToByte(cleanedCommand);
return result != null ? result : new byte[0];
} catch (Exception e) {
// 如果标准方法失败，手动转换为十六进制字节数组
try {
String cleanedCommand = command.trim()
.replaceAll("\\s+", "")
.replaceAll("(?i)^0x", "");
if (cleanedCommand.length() % 2 != 0) {
cleanedCommand = "0" + cleanedCommand;
}
// 手动将每两个字符转换为一个字节
byte[] result = new byte[cleanedCommand.length() / 2];
for (int i = 0; i < cleanedCommand.length(); i += 2) {
String hexByte = cleanedCommand.substring(i, i + 2);
result[i / 2] = (byte) Integer.parseInt(hexByte, 16);
}
return result;
} catch (Exception e2) {
// 如果手动转换也失败，返回空数组
return new byte[0];
}
}
}
public static byte getByte(byte[] src, int offset) {
return src[offset];
}

293
fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java Normal file
View File

@ -0,0 +1,293 @@
package com.fastbee.iot.data.aspect;
import com.fastbee.common.core.device.DeviceAndProtocol;
import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.core.mq.message.DeviceData;
import com.fastbee.common.core.mq.message.DeviceMessage;
import com.fastbee.common.enums.DeviceLogTypeEnum;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.domain.DeviceLog;
import com.fastbee.iot.service.IDeviceService;
import com.fastbee.iot.service.IProductService;
import com.fastbee.iot.tdengine.service.ILogService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* 设备调试错误日志AOP切面
* 用于拦截数据解析和指令发送的异常，自动记录错误日志
*
* @author fastbee
* @date 2025-01-08
*/
@Aspect
@Component
@Order(100)
@Slf4j
public class DeviceDebugLogAspect {
@Resource
private ILogService logService;
@Resource
private IDeviceService deviceService;
@Resource
private IProductService productService;
/**
* 拦截数据解析方法
*/
@Pointcut("execution(* com.fastbee.iot.data.service.impl.DeviceReportMessageServiceImpl.parseReportMsg(..))")
public void parseReportMsgPointcut() {
}
/**
* 拦截指令发送方法
*/
@Pointcut("execution(* com.fastbee.iot.data.service.impl.DeviceMessageServiceImpl.messagePost(..))")
public void messagePostPointcut() {
}
/**
* 拦截协议解析方法
*/
@Pointcut("execution(* com.fastbee.protocol.base.protocol.IProtocol.decode(..))")
public void protocolDecodePointcut() {
}
/**
* 拦截数据解析异常
*/
@AfterThrowing(pointcut = "parseReportMsgPointcut()", throwing = "e")
public void afterThrowingParseReportMsg(JoinPoint joinPoint, Throwable e) {
try {
Object[] args = joinPoint.getArgs();
if (args != null && args.length > 0 && args[0] instanceof DeviceReportBo) {
DeviceReportBo bo = (DeviceReportBo) args[0];
String serialNumber = bo.getSerialNumber();
if (StringUtils.isEmpty(serialNumber)) {
log.warn("设备编号为空，跳过错误日志记录");
return;
}
// 处理原始数据，可能是byte[]或String
String sourceData = "";
try {
if (bo.getData() != null && bo.getData().length > 0) {
sourceData = ByteBufUtil.hexDump(bo.getData());
} else if (StringUtils.isNotEmpty(bo.getSources())) {
sourceData = bo.getSources();
}
} catch (Exception ex) {
log.warn("处理原始数据失败: {}", ex.getMessage());
sourceData = bo.getSources() != null ? bo.getSources() : "";
}
String errorMsg = e.getMessage();
if (StringUtils.isEmpty(errorMsg)) {
errorMsg = e.getClass().getName();
}
// 直接从bo中获取协议编码，避免额外的查询
String protocolCode = bo.getProtocolCode();
saveErrorLog(serialNumber, sourceData, errorMsg, protocolCode, bo.getTopicName(),
DeviceLogTypeEnum.DATA_PARSE_ERROR.getType(), "PARSE_ERROR");
}
} catch (Exception ex) {
log.error("记录数据解析错误日志失败: {}", ex.getMessage(), ex);
}
}
/**
* 拦截指令发送异常
*/
@AfterThrowing(pointcut = "messagePostPointcut()", throwing = "e")
public void afterThrowingMessagePost(JoinPoint joinPoint, Throwable e) {
try {
Object[] args = joinPoint.getArgs();
if (args != null && args.length > 0 && args[0] instanceof DeviceMessage) {
DeviceMessage deviceMessage = (DeviceMessage) args[0];
String serialNumber = deviceMessage.getSerialNumber();
if (StringUtils.isEmpty(serialNumber)) {
log.warn("设备编号为空，跳过错误日志记录");
return;
}
// 处理原始数据
String sourceData = "";
try {
if (deviceMessage.getMessage() != null) {
Object msg = deviceMessage.getMessage();
if (msg instanceof byte[]) {
sourceData = ByteBufUtil.hexDump((byte[]) msg);
} else {
sourceData = msg.toString();
}
}
} catch (Exception ex) {
log.warn("处理原始数据失败: {}", ex.getMessage());
}
String errorMsg = e.getMessage();
if (StringUtils.isEmpty(errorMsg)) {
errorMsg = e.getClass().getName();
}
// 获取协议编码（通过设备编号查询，有缓存）
String protocolCode = getProtocolCodeBySerialNumber(serialNumber);
saveErrorLog(serialNumber, sourceData, errorMsg, protocolCode, deviceMessage.getTopicName(),
DeviceLogTypeEnum.COMMAND_SEND_ERROR.getType(), "SEND_ERROR");
}
} catch (Exception ex) {
log.error("记录指令发送错误日志失败: {}", ex.getMessage(), ex);
}
}
/**
* 拦截协议解析异常
*/
@AfterThrowing(pointcut = "protocolDecodePointcut()", throwing = "e")
public void afterThrowingProtocolDecode(JoinPoint joinPoint, Throwable e) {
try {
Object[] args = joinPoint.getArgs();
if (args != null && args.length > 0 && args[0] instanceof DeviceData) {
DeviceData deviceData =
(DeviceData) args[0];
String serialNumber = deviceData.getSerialNumber();
if (StringUtils.isEmpty(serialNumber)) {
log.warn("设备编号为空，跳过错误日志记录");
return;
}
// 处理原始数据
String sourceData = "";
try {
ByteBuf buf = deviceData.getBuf();
if (buf != null) {
sourceData = ByteBufUtil.hexDump(buf);
} else if (deviceData.getData() != null && deviceData.getData().length > 0) {
sourceData = ByteBufUtil.hexDump(deviceData.getData());
}
} catch (Exception ex) {
log.warn("处理原始数据失败: {}", ex.getMessage());
}
String errorMsg = e.getMessage();
if (StringUtils.isEmpty(errorMsg)) {
errorMsg = e.getClass().getName();
}
// 获取协议编码：优先从productId获取（有缓存），否则通过设备编号查询
String protocolCode = null;
if (deviceData.getProductId() != null) {
protocolCode = getProtocolCodeByProductId(deviceData.getProductId());
}
if (StringUtils.isEmpty(protocolCode) && StringUtils.isNotEmpty(serialNumber)) {
protocolCode = getProtocolCodeBySerialNumber(serialNumber);
}
// 协议解析错误也归类为数据解析错误
saveErrorLog(serialNumber, sourceData, errorMsg, protocolCode, deviceData.getTopicName(),
DeviceLogTypeEnum.DATA_PARSE_ERROR.getType(), "PARSE_ERROR");
}
} catch (Exception ex) {
log.error("记录协议解析错误日志失败: {}", ex.getMessage(), ex);
}
}
/**
* 根据设备编号获取协议编码（有缓存）
*/
private String getProtocolCodeBySerialNumber(String serialNumber) {
try {
DeviceAndProtocol deviceAndProtocol =
deviceService.selectProtocolBySerialNumber(serialNumber);
if (deviceAndProtocol != null) {
return deviceAndProtocol.getProtocolCode();
}
} catch (Exception ex) {
log.warn("获取协议编码失败: {}, 设备编号: {}", ex.getMessage(), serialNumber);
}
return null;
}
/**
* 根据产品ID获取协议编码（有缓存）
*/
private String getProtocolCodeByProductId(Long productId) {
try {
if (productId != null) {
return productService.getProtocolByProductId(productId);
}
} catch (Exception ex) {
log.warn("根据产品ID获取协议编码失败: {}, 产品ID: {}", ex.getMessage(), productId);
}
return null;
}
/**
* 保存错误日志到iot_device_log表
*/
private void saveErrorLog(String serialNumber, String sourceData, String errorMsg,
String protocolCode, String topicName, Integer logType, String identity) {
try {
if (StringUtils.isEmpty(serialNumber)) {
log.warn("设备编号为空，无法保存错误日志");
return;
}
// 获取设备信息
Device device = null;
try {
device = deviceService.selectDeviceBySerialNumber(serialNumber);
} catch (Exception ex) {
log.warn("获取设备信息失败: {}, 设备编号: {}", ex.getMessage(), serialNumber);
}
if (device == null) {
log.warn("设备不存在，无法保存错误日志: {}", serialNumber);
return;
}
DeviceLog deviceLog = new DeviceLog();
deviceLog.setSerialNumber(serialNumber);
deviceLog.setDeviceId(device.getDeviceId());
deviceLog.setDeviceName(device.getDeviceName());
deviceLog.setLogType(logType);
// 存储原始数据包（限制64字符）
deviceLog.setLogValue(StringUtils.substring(sourceData, 0, 64));
// 存储错误信息（限制200字符）
deviceLog.setRemark(StringUtils.substring(errorMsg, 0, 500));
deviceLog.setIdentity(identity);
deviceLog.setModelName(logType == DeviceLogTypeEnum.DATA_PARSE_ERROR.getType() ?
"数据解析错误" : "指令发送错误");
deviceLog.setCreateTime(DateUtils.getNowDate());
deviceLog.setUserId(device.getTenantId());
deviceLog.setUserName(device.getTenantName());
deviceLog.setTenantId(device.getTenantId());
deviceLog.setTenantName(device.getTenantName());
deviceLog.setIsMonitor(0);
deviceLog.setMode(3); // 3=其他
logService.saveDeviceLog(deviceLog);
log.info("已保存错误日志到数据库 - 设备: {}, 错误类型: {}, 错误信息: {}",
serialNumber, identity, StringUtils.substring(errorMsg, 0, 50));
} catch (Exception ex) {
log.error("保存错误日志失败: {}", ex.getMessage(), ex);
}
}
}

8
fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java
View File

@ -15,6 +15,7 @@ import com.fastbee.common.enums.TopicType;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.common.utils.gateway.protocol.ByteUtils;
import com.fastbee.iot.data.service.impl.MessageManager;
import com.fastbee.iot.domain.DeviceJob;
import com.fastbee.iot.domain.ModbusJob;
@ -81,6 +82,7 @@ public class DevicePropFetchConsumer {
public void consume(ModbusPollJob job) {
Integer type = job.getType();
DeviceJob deviceJob = job.getDeviceJob();
// type 1:手动下发 2:定时任务执行
ModbusPollMsg task = type == 1 ? job.getPollMsg() : getCommandList(deviceJob);
if (CollectionUtils.isEmpty(task.getCommandList()))return;
String serialNumber = task.getSerialNumber();
@ -96,17 +98,19 @@ public class DevicePropFetchConsumer {
for (String command : commandList) {
String cacheKey = RedisKeyBuilder.buildModbusRuntimeCacheKey(serialNumber);
redisCache.zSetAdd(cacheKey, command, DateUtils.getTimestampSeconds());
// 不进行格式校验，直接转换数据
byte[] commandData = ByteUtils.convertToHexByteArray(command);
switch (serverType) {
//通过mqtt内部客户端 下发指令
case MQTT:
publish(topic, ByteBufUtil.decodeHexDump(command));
publish(topic, commandData);
log.info("=>MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic);
break;
// 下发TCP客户端
case TCP:
Message msg = new Message();
msg.setClientId(serialNumber);
msg.setPayload(Unpooled.wrappedBuffer(ByteBufUtil.decodeHexDump(command)));
msg.setPayload(Unpooled.wrappedBuffer(commandData));
messageManager.requestR(serialNumber, msg, Message.class);
log.info("=>TCP-线程=[{}],轮询指令:[{}]", Thread.currentThread().getName(), command);
break;

1
fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/subscribeCallback.java
View File

@ -6,6 +6,7 @@ import com.fastbee.common.enums.TopicType;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.common.utils.gateway.protocol.ByteUtils;
import com.fastbee.ruleEngine.context.MsgContext;
import com.fastbee.iot.ruleEngine.RuleProcess;
import com.fastbee.mqttclient.IEmqxMessageProducer;

16
fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java
View File

@ -19,6 +19,7 @@ import com.fastbee.common.utils.BitUtils;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.gateway.CRC16Utils;
import com.fastbee.common.utils.gateway.protocol.ByteUtils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.common.utils.modbus.ModbusUtils;
import com.fastbee.iot.cache.ITSLCache;
@ -113,9 +114,22 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
break;
case PROPERTY_POST:
//下发的不经过mqtt或TCP直接转发到数据处理模块
// 强制按十六进制处理，直接转发原始数据
byte[] messageData;
Object message = deviceMessage.getMessage();
if (message == null) {
messageData = new byte[0];
} else if (message instanceof byte[]) {
messageData = (byte[]) message;
} else if (message instanceof String) {
messageData = ByteUtils.convertToHexByteArray((String) message);
} else {
// 其他类型直接转换为字符串后按十六进制处理
messageData = ByteUtils.convertToHexByteArray(message.toString());
}
DeviceReportBo reportBo = DeviceReportBo.builder()
.serverType(ServerType.explain(transport))
.data(BitUtils.hexStringToByteArray(deviceMessage.getMessage().toString()))
.data(messageData)
.platformDate(DateUtils.getNowDate())
.serialNumber(serialNumber)
.topicName(topicName).build();

10
fastbee-protocol/fastbee-protocol-base/src/main/java/com/fastbee/protocol/service/impl/ProtocolManagerServiceImpl.java
View File

@ -15,7 +15,9 @@ import com.fastbee.protocol.domain.DeviceProtocol;
import com.fastbee.protocol.service.IProtocolManagerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.formula.functions.T;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@ -52,7 +54,9 @@ public class ProtocolManagerServiceImpl<M> implements IProtocolManagerService {
protocol.setDelFlag(0);
List<Protocol> protocolList = protocolService.selectByCondition(protocol);
annotations.forEach((key,value)->{
SysProtocol annotation = value.getClass().getAnnotation(SysProtocol.class);
// 获取目标类而非代理类，避免AOP代理导致注解丢失
Class<?> targetClass = AopUtils.getTargetClass(value);
SysProtocol annotation = AnnotationUtils.findAnnotation(targetClass, SysProtocol.class);
ProtocolDto protocolDto = new ProtocolDto();
protocolDto.setCode(annotation.protocolCode());
protocolDto.setName(annotation.name());
@ -97,7 +101,9 @@ public class ProtocolManagerServiceImpl<M> implements IProtocolManagerService {
Map<String, IProtocol> annotations = SpringUtils.getBeanWithAnnotation(SysProtocol.class);
IProtocol protocol = null;
for (IProtocol item : annotations.values()) {
SysProtocol annotation = item.getClass().getAnnotation(SysProtocol.class);
// 获取目标类而非代理类，避免AOP代理导致注解丢失
Class<?> targetClass = AopUtils.getTargetClass(item);
SysProtocol annotation = AnnotationUtils.findAnnotation(targetClass, SysProtocol.class);
protocolMap.put(annotation.protocolCode(), item);
if (annotation.protocolCode().equals(protocolCode)){
protocol = item;

2
fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/MessageAdapter.java
View File

@ -51,7 +51,7 @@ public class MessageAdapter implements MessageDecoder, MessageEncoder{
Long productId = null;
String devNum = new String(ByteBufUtil.getBytes(buf));
String dump = ByteBufUtil.hexDump(buf);
log.info("=>上报hex数据:{}", dump);
log.info("{}=>上报hex数据:{}",clientId, dump);
//这里兼容一下TCP整包发送的数据(整包==设备编号，数据一起发送)
if (clientId == null){