opensource/Universal-IoT-Java
3
0
Fork 0
mirror of https://gitee.com/NexIoT/Universal-IoT-Java.git synced 2026-10-31 02:52:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

news：合并2.0商业版代码：framework模块

Browse Source 
版本更新汇总 (2025-08-10 至 2025-09-23)
核心功能优化:
- 新增UDP模块并默认开启，TCP基础上增加UDP适配
- 支持网关绑定多产品子设备，优化设备重连机制
- 新增Ezviz（萤石）接入模块，调整imoulife支持自有平台账号

性能与稳定性:
- 优化设备影子热数据存储，增加强制刷盘功能
- 产品列表及MQTT查询增加多级缓存策略
- 强化TCP安全机制，新增异常攻击封禁策略

功能完善:
- 集成rulego可视化规则引擎基础框架
- Web-IDE新增20+ modbus/hex封装方法
- 优化乐城播放器界面及TCP连接信息展示

问题修复:
- 解决系统间强依赖及注入相关问题
- 修复时间格式化、设备创建时间显示问题
- 完善日志处理及应用推送相关异常

其他改进:
- 重构TCP粘包处理，删除冗余代码
- 增加Docker部署配置
- 细化权限控制策略
This commit is contained in:
奈科斯（NexIoT）
2025-09-23 11:04:19 +08:00
parent 7aef1ba8e3
commit ce3d4f50bb
22 changed files with 250 additions and 552 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cn-universal-framework/cn-universal-common/src/main/java/cn/universal/common/constant/IoTConstant.java
+122 -7
View File
@@ -23,7 +23,8 @@ import java.util.stream.Stream;
/** IoT常量 */
public interface IoTConstant {
/*发送直通第三方平台*/
String DOWN_TO_THIRD_PLATFORM = "downToThirdPlatform";
String CURRENT_INSTANCE_ID = "instanceId";
public final String CERT_DEFAULT_KEY = "default-tcp";
public final String HTTP_UP_BLACK_LIST = "httpUpBlackList";
@@ -126,6 +127,7 @@ public interface IoTConstant {
public static final String TRACE_ID = "traceId";
static final String defaultMetadata =
"""
{"tags":[],"events":[{"id":"online","name":"上线","valueType":{"type":"string"}},{"id":"offline","name":"下线","valueType":{"type":"string"}}],"functions":[],"properties":[]}""";
@@ -134,9 +136,11 @@ public interface IoTConstant {
ctaiot,
ezviz,
onenet,
lechen,
imoulife,
tcp,
sniTcp
udp,
mqtt,
http
}
/** 非设备真实上报事件 */
@@ -203,8 +207,109 @@ public interface IoTConstant {
}
}
/** 下行指令 */
/** 下行指令 - 重新设计为更通俗易懂的指令 */
enum DownCmd {
// 设备管理指令
DEVICE_ADD("设备添加"),
DEVICE_DELETE("设备删除"),
DEVICE_UPDATE("设备更新"),
DEVICE_INFO("设备信息查询"),
DEVICE_STATUS("设备状态查询"),
DEVICE_ONLINE_CHECK("设备在线检查"),
// 摄像头控制指令
CAMERA_TURN("摄像头转动"),
CAMERA_PTZ_CONTROL("云台控制"),
CAMERA_SNAPSHOT("摄像头截图"),
CAMERA_LIVE_STREAM("摄像头直播"),
CAMERA_PLAYBACK("摄像头回放"),
CAMERA_RECORD_START("开始录像"),
CAMERA_RECORD_STOP("停止录像"),
CAMERA_FLIP_SET("画面翻转设置"),
// 存储相关物模型函数
STORAGE_LOCAL_STREAM_SET_FUNC("storageLocalStreamSet"),
STORAGE_LOCAL_STREAM_QUERY_FUNC("storageLocalStreamQuery"),
STORAGE_LOCAL_PLAN_SET_FUNC("storageLocalPlanSet"),
STORAGE_LOCAL_PLAN_QUERY_FUNC("storageLocalPlanQuery"),
STORAGE_CLOUD_RECORDS_QUERY_FUNC("storageCloudRecordsQuery"),
STORAGE_CLOUD_UNUSED_LIST_FUNC("storageCloudUnusedList"),
STORAGE_CLOUD_CALL_COUNT_QUERY_FUNC("storageCloudCallCountQuery"),
STORAGE_CLOUD_SERVICE_SET_FUNC("storageCloudServiceSet"),
STORAGE_CLOUD_LIST_QUERY_FUNC("storageCloudListQuery"),
STORAGE_CLOUD_UNBIND_FUNC("storageCloudUnbind"),
STORAGE_CLOUD_OPEN_FUNC("storageCloudOpen"),
STORAGE_FREE_CLOUD_SET_FUNC("storageFreeCloudSet"),
STORAGE_SDCARD_FORMAT_FUNC("storageSdcardFormat"),
STORAGE_SDCARD_INFO_GET_FUNC("storageSdcardInfoGet"),
STORAGE_SDCARD_STATUS_GET_FUNC("storageSdcardStatusGet"),
STORAGE_CLOUD_VIDEO_COUNT_QUERY_FUNC("storageCloudVideoCountQuery"),
STORAGE_LOCAL_VIDEO_COUNT_QUERY_FUNC("storageLocalVideoCountQuery"),
VIDEO_DOWNLOAD_FUNC("videoDownload"),
// 设备相关物模型函数
DEVICE_SOUND_VOLUME_GET_FUNC("deviceSoundVolumeGet"),
DEVICE_SOUND_VOLUME_SET_FUNC("deviceSoundVolumeSet"),
DEVICE_CAMERA_STATUS_GET_FUNC("deviceCameraStatusGet"),
DEVICE_WIFI_SET_FUNC("deviceWifiSet"),
DEVICE_WIFI_SCAN_FUNC("deviceWifiScan"),
DEVICE_UPGRADE_FUNC("deviceUpgrade"),
DEVICE_VERSION_QUERY_FUNC("deviceVersionQuery"),
DEVICE_CLOUD_INFO_GET_FUNC("deviceCloudInfoGet"),
DEVICE_ENABLE_SET_FUNC("deviceEnableSet"),
DEVICE_RESTART_FUNC("deviceRestart"),
// 报警相关物模型函数
ALARM_MESSAGE_QUERY_FUNC("alarmMessageQuery"),
// 存储管理指令
STORAGE_INFO("存储信息查询"),
STORAGE_FORMAT("存储格式化"),
STORAGE_CLOUD_ENABLE("云存储开启"),
STORAGE_CLOUD_DISABLE("云存储关闭"),
STORAGE_LOCAL_ENABLE("本地存储开启"),
STORAGE_LOCAL_DISABLE("本地存储关闭"),
STORAGE_LOCAL_STREAM_SET("设置本地录像视频流"),
STORAGE_LOCAL_STREAM_QUERY("查询本地录像视频流"),
STORAGE_LOCAL_PLAN_SET("设置本地录像计划"),
STORAGE_LOCAL_PLAN_QUERY("查询本地录像计划"),
STORAGE_CLOUD_RECORDS_QUERY("查询云录像片段"),
STORAGE_CLOUD_UNUSED_LIST("获取未启用的云存储服务"),
STORAGE_CLOUD_CALL_COUNT_QUERY("查询云存储开通接口剩余调用次数"),
STORAGE_CLOUD_SERVICE_SET("设置云存储服务开关"),
STORAGE_CLOUD_LIST_QUERY("查询设备云存储服务"),
STORAGE_CLOUD_UNBIND("解绑设备云存储"),
STORAGE_CLOUD_OPEN("开通设备云存储"),
STORAGE_FREE_CLOUD_SET("设置免费云存储服务"),
STORAGE_SDCARD_FORMAT("格式化SD卡"),
STORAGE_SDCARD_INFO_GET("获取SD卡信息"),
STORAGE_SDCARD_STATUS_GET("获取SD卡状态"),
STORAGE_CLOUD_VIDEO_COUNT_QUERY("查询云录像数量"),
STORAGE_LOCAL_VIDEO_COUNT_QUERY("查询本地录像数量"),
VIDEO_DOWNLOAD("下载录像"),
// 设备设置指令
DEVICE_RESTART("设备重启"),
DEVICE_UPGRADE("设备升级"),
DEVICE_WIFI_SET("WiFi设置"),
DEVICE_WIFI_SCAN("WiFi扫描"),
DEVICE_SOUND_VOLUME_GET("获取设备音量"),
DEVICE_SOUND_VOLUME_SET("设置设备音量"),
DEVICE_CAMERA_STATUS_GET("获取摄像头状态"),
DEVICE_VERSION_QUERY("查询设备版本信息"),
DEVICE_CLOUD_INFO_GET("获取设备云存储信息"),
DEVICE_ENABLE_SET("设置设备使能开关"),
// 报警管理指令
ALARM_MESSAGE_QUERY("查询报警消息"),
// 产品管理指令
PRODUCT_ADD("产品添加"),
PRODUCT_UPDATE("产品更新"),
PRODUCT_DELETE("产品删除"),
PRODUCT_PUBLISH("产品发布"),
// 兼容旧版本指令（保留）
DEV_ADD,
DEV_ADDS,
DEV_FUNCTION,
@@ -214,7 +319,6 @@ public interface IoTConstant {
MANUAL_CAPTURE,
DEV_CONTROLLING,
DEV_MONITOR_PLAY,
DEV_ELECTRIC_QUANTITY,
DEV_DOOR_KEYS,
DEV_OPENDOOR_RECORD,
@@ -264,7 +368,6 @@ public interface IoTConstant {
FRAME_REVERSE_STATUS,
SOUND_VOLUME_SIZE_GET,
CAMERA_STATUS,
CAMERA_SNAPSHOT,
CONTROL_DEVICE_WIFI,
WIFI_AROUND,
UPGRADE_DEVICE,
@@ -277,7 +380,15 @@ public interface IoTConstant {
ALARM_MESSAGE,
DEV_MONITOR_CHECK_ONLINE;
private DownCmd() {}
private String description;
DownCmd() {
this.description = super.toString().toLowerCase();
}
DownCmd(String description) {
this.description = description;
}
@JsonCreator
public static DownCmd find(String value) {
@@ -291,6 +402,10 @@ public interface IoTConstant {
public String getValue() {
return super.toString().toLowerCase();
}
public String getDescription() {
return description;
}
}
/** 设备订阅 */
cn-universal-framework/cn-universal-common/src/main/java/cn/universal/common/event/EventProcessorFactory.java
+23 -18
View File
@@ -1,8 +1,11 @@
package cn.universal.common.event;
import cn.hutool.core.map.MapUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.universal.common.event.processer.FenceEventProcessor;
import cn.universal.common.event.processer.ProductConfigProcessor;
import cn.universal.common.event.processer.TcpDownProcessor;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -26,10 +29,6 @@ public class EventProcessorFactory {
@Autowired(required = false)
private TcpDownProcessor tcpDownProcessor;
// 产品配置处理器
@Autowired(required = false)
private ProductConfigProcessor productConfigProcessor;
/** 处理电子围栏事件 */
public void handleFenceEvent(EventMessage message) {
if (fenceEventProcessor != null) {
@@ -58,27 +57,33 @@ public class EventProcessorFactory {
/** 处理产品配置更新事件 */
public void handleProductConfigUpdated(EventMessage message) {
if (productConfigProcessor != null) {
try {
productConfigProcessor.handleProductConfigUpdated(message);
} catch (Exception e) {
log.error("[事件处理器] 产品配置更新处理失败", e);
try {
Map<String, ProductConfigProcessor> beans =
SpringUtil.getBeansOfType(ProductConfigProcessor.class);
if (MapUtil.isNotEmpty(beans)) {
beans.forEach(
(k, v) -> {
v.handleProductConfigUpdated(message);
});
}
} else {
log.warn("[事件处理器] ProductConfigProcessor未找到，跳过产品配置更新处理");
} catch (Exception e) {
log.error("[事件处理器] 产品配置更新处理失败", e);
}
}
/** 处理产品配置更新事件 */
public void handleProtocolUpdated(EventMessage message) {
if (productConfigProcessor != null) {
try {
productConfigProcessor.handleProtocolUpdated(message);
} catch (Exception e) {
log.error("[事件处理器] 产品协议更新处理失败", e);
try {
Map<String, ProductConfigProcessor> beans =
SpringUtil.getBeansOfType(ProductConfigProcessor.class);
if (MapUtil.isNotEmpty(beans)) {
beans.forEach(
(k, v) -> {
v.handleProductConfigUpdated(message);
});
}
} else {
log.warn("[事件处理器] ProductConfigProcessor未找到，产品协议更新处理失败");
} catch (Exception e) {
log.error("[事件处理器] 产品协议更新处理失败", e);
}
}
}
cn-universal-framework/cn-universal-common/src/main/java/cn/universal/common/event/EventTopics.java
+15
View File
@@ -35,6 +35,11 @@ public class EventTopics {
/** TCP下行指令事件（模式匹配） 用于处理TCP下行指令 */
public static final String TCP_DOWN = "tcp:down:*";
// ==================== UDP相关事件 ====================
/** UDP下行指令事件（模式匹配） 用于处理UDP下行指令 */
public static final String UDP_DOWN = "udp:down:*";
// ==================== 产品配置相关事件 ====================
/** 产品配置更新事件 用于通知产品配置变更 */
@@ -56,4 +61,14 @@ public class EventTopics {
public static String getTcpDownTopic(String instanceId) {
return "tcp:down:" + instanceId;
}
/**
* 获取UDP下行指令主题（带实例ID）
*
* @param instanceId 实例ID
* @return 完整的UDP下行指令主题
*/
public static String getUdpDownTopic(String instanceId) {
return "udp:down:" + instanceId;
}
}
cn-universal-framework/cn-universal-common/src/main/java/cn/universal/common/event/RedisEventPublisher.java
+1 -1
View File
@@ -42,7 +42,7 @@ public class RedisEventPublisher implements EventPublisher {
String message = JSONUtil.toJsonStr(eventMessage);
redisTemplate.convertAndSend(topic, message);
log.debug("[Redis事件发布] 发布事件到主题: {}, 消息: {}", topic, message);
log.info("[Redis事件发布] 发布事件到主题: {}, 消息: {}", topic, message);
} catch (Exception e) {
log.error("[Redis事件发布] 发布事件失败: topic={}, event={}", topic, event, e);
}
cn-universal-framework/cn-universal-common/src/main/java/cn/universal/common/event/RedisEventSubscriber.java
+4 -1
View File
@@ -154,7 +154,10 @@ public class RedisEventSubscriber {
private boolean isOwnMessage(EventMessage message) {
try {
String nodeId = message.getNodeId();
return instanceIdProvider.getInstanceId().equals(nodeId);
String instanceId = instanceIdProvider.getInstanceId();
final boolean b = instanceId.equalsIgnoreCase(nodeId);
log.info("判断是否自己的消息={},nodeId={}, instanceId={}", b, nodeId, instanceId);
return b;
} catch (Exception e) {
return false;
}
cn-universal-framework/cn-universal-common/src/main/java/cn/universal/common/exception/IoTErrorCode.java
+1 -1
View File
@@ -59,7 +59,7 @@ public enum IoTErrorCode {
APPLICATION_NOT_FOR_YOU(801, "应用没有操作权限！"),
/** 产品 */
PRODUCT_NOT_FOR_YOU(901, "您没有权限操作此产品！"),
PRODUCT_NOT_FOR_YOU(901, "无产品权限"),
PRODUCT_NOT_EXIST(902, "产品不存在");
private Integer code;
cn-universal-framework/cn-universal-core/pom.xml
-12
View File
@@ -46,12 +46,6 @@
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
</dependency>
<!-- Redis 相关依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- GraalJS - 替代 Nashorn -->
<dependency>
<groupId>org.graalvm.js</groupId>
@@ -70,12 +64,6 @@
<artifactId>spring-security-crypto</artifactId>
</dependency>
<!-- <dependency>-->
<!-- <groupId>org.apache.commons</groupId>-->
<!-- <artifactId>commons-pool2</artifactId>-->
<!-- </dependency>-->
<!-- Lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/engine/functions/IotRedisUtil.java
-303
View File
@@ -1,303 +0,0 @@
/*
*
* Copyright (c) 2025, IoT-Universal. All Rights Reserved.
*
* @Description: 本文件由 Aleo 开发并拥有版权，未经授权严禁擅自商用、复制或传播。
* @Author: Aleo
* @Email: wo8335224@gmail.com
* @Wechat: outlookFil
*
*
*/
package cn.universal.core.engine.functions;
import cn.universal.core.engine.annotation.Comment;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class IotRedisUtil {
private final StringRedisTemplate redisTemplate;
private static final String DEVICE_PREFIX = "univiot:device:";
public IotRedisUtil(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
}
// ==================== 基础数据结构操作（带过期时间）====================
@Comment("设置字符串值（带过期时间）")
public void setString(String key, String value, long timeout, String unit) {
try {
redisTemplate.opsForValue().set(key, value, timeout, TimeUnit.valueOf(unit));
} catch (RedisConnectionFailureException e) {
log.error("Redis连接失败 - Key: {}, Error: {}", key, e.getMessage());
} catch (RedisSystemException e) {
log.error("Redis系统异常 - Key: {}, Error: {}", key, e.getCause().getMessage());
} catch (Exception e) {
log.error("设置字符串值异常 - Key: {}, Error: {}", key, e.getMessage());
}
}
@Comment("设置Hash字段值（带过期时间）")
public void setHashField(String key, String field, String value, long timeout, String unit) {
try {
redisTemplate.opsForHash().put(key, field, value);
redisTemplate.expire(key, timeout, TimeUnit.valueOf(unit));
} catch (RedisConnectionFailureException e) {
log.error("Redis连接失败 - Key: {}, Field: {}", key, field);
} catch (Exception e) {
log.error("设置Hash字段异常 - Key: {}, Field: {}, Error: {}", key, field, e.getMessage());
}
}
@Comment("从左侧插入List（带过期时间）")
public Long leftPush(String key, String value, long timeout, String unit) {
try {
Long result = redisTemplate.opsForList().leftPush(key, value);
redisTemplate.expire(key, timeout, TimeUnit.valueOf(unit));
return result;
} catch (RedisConnectionFailureException e) {
log.error("Redis连接失败 - Key: {}", key);
return 0L;
} catch (Exception e) {
log.error("List左插入异常 - Key: {}, Error: {}", key, e.getMessage());
return 0L;
}
}
@Comment("添加Set成员（带过期时间）")
public Long addToSet(String key, String value, long timeout, String unit) {
try {
Long result = redisTemplate.opsForSet().add(key, value);
redisTemplate.expire(key, timeout, TimeUnit.valueOf(unit));
return result;
} catch (RedisConnectionFailureException e) {
log.error("Redis连接失败 - Key: {}", key);
return 0L;
} catch (Exception e) {
log.error("Set添加成员异常 - Key: {}, Error: {}", key, e.getMessage());
return 0L;
}
}
@Comment("添加ZSet成员（带过期时间）")
public Boolean addToZSet(String key, String value, double score, long timeout, String unit) {
try {
Boolean result = redisTemplate.opsForZSet().add(key, value, score);
redisTemplate.expire(key, timeout, TimeUnit.valueOf(unit));
return result;
} catch (RedisConnectionFailureException e) {
log.error("Redis连接失败 - Key: {}", key);
return false;
} catch (Exception e) {
log.error("ZSet添加成员异常 - Key: {}, Error: {}", key, e.getMessage());
return false;
}
}
// ==================== 物联网场景专用方法 ====================
@Comment("更新设备状态（自动续期）")
public void updateDeviceStatus(
String deviceId, String field, String value, long timeout, String unit) {
String key = DEVICE_PREFIX + "status:" + deviceId;
try {
redisTemplate.opsForHash().put(key, field, value);
redisTemplate.expire(key, timeout, TimeUnit.valueOf(unit));
} catch (RedisConnectionFailureException e) {
log.error("设备状态更新失败 - Device: {}, Field: {}", deviceId, field);
} catch (Exception e) {
log.error("更新设备状态异常 - Device: {}, Field: {}, Error: {}", deviceId, field, e.getMessage());
}
}
@Comment("获取设备状态")
public Map<String, String> getDeviceStatus(String deviceId) {
String key = DEVICE_PREFIX + "status:" + deviceId;
try {
Map<Object, Object> rawMap = redisTemplate.opsForHash().entries(key);
return convertToStringMap(rawMap);
} catch (RedisConnectionFailureException e) {
log.error("获取设备状态失败 - Device: {}", deviceId);
return Collections.emptyMap();
} catch (Exception e) {
log.error("获取设备状态异常 - Device: {}, Error: {}", deviceId, e.getMessage());
return Collections.emptyMap();
}
}
@Comment("推送设备指令到队列（带过期时间）")
public void pushDeviceCommand(
String deviceId, String commandType, String params, long timeout, String unit) {
String key = DEVICE_PREFIX + "cmd:" + deviceId;
try {
String command =
String.format(
"{\"type\":\"%s\",\"params\":%s,\"timestamp\":%d}",
commandType, params, System.currentTimeMillis());
// 使用管道操作确保原子性
redisTemplate.executePipelined(
(RedisCallback<Object>)
connection -> {
connection.rPush(key.getBytes(), command.getBytes());
connection.expire(key.getBytes(), TimeUnit.valueOf(unit).toSeconds(timeout));
return null;
});
} catch (RedisConnectionFailureException e) {
log.error("指令推送失败 - Device: {}, Command: {}", deviceId, commandType);
} catch (Exception e) {
log.error(
"指令推送异常 - Device: {}, Command: {}, Error: {}", deviceId, commandType, e.getMessage());
}
}
@Comment("获取并移除下一个指令")
public String popNextCommand(String deviceId) {
String key = DEVICE_PREFIX + "cmd:" + deviceId;
try {
return redisTemplate.opsForList().leftPop(key);
} catch (RedisConnectionFailureException e) {
log.error("获取指令失败 - Device: {}", deviceId);
return null;
} catch (Exception e) {
log.error("获取指令异常 - Device: {}, Error: {}", deviceId, e.getMessage());
return null;
}
}
@Comment("记录设备位置")
public void recordDeviceLocation(String deviceId, double lng, double lat) {
try {
redisTemplate
.opsForGeo()
.add(
DEVICE_PREFIX + "geo:devices",
new RedisGeoCommands.GeoLocation<>(deviceId, new Point(lng, lat)));
} catch (RedisConnectionFailureException e) {
log.error("位置记录失败 - Device: {}", deviceId);
} catch (Exception e) {
log.error("位置记录异常 - Device: {}, Error: {}", deviceId, e.getMessage());
}
}
@Comment("记录设备事件")
public void recordDeviceEvent(String deviceId, String eventType, Map<String, String> eventData) {
String key = DEVICE_PREFIX + "events:" + deviceId;
try {
Map<String, String> fields = new HashMap<>();
fields.put("eventType", eventType);
fields.put("timestamp", String.valueOf(System.currentTimeMillis()));
fields.putAll(eventData);
redisTemplate.opsForStream().add(key, fields);
} catch (RedisConnectionFailureException e) {
log.error("事件记录失败 - Device: {}, Event: {}", deviceId, eventType);
logToLocalStorage(eventData); // 降级到本地存储
} catch (Exception e) {
log.error("事件记录异常 - Device: {}, Event: {}, Error: {}", deviceId, eventType, e.getMessage());
logToLocalStorage(eventData); // 降级到本地存储
}
}
@Comment("缓存设备数据并记录历史")
public void cacheDeviceData(
String deviceId, String dataType, String value, long timeout, String unit) {
String currentKey = DEVICE_PREFIX + "data:current:" + deviceId + ":" + dataType;
try {
// 获取上一次值
String lastValue = redisTemplate.opsForValue().get(currentKey);
// 存储历史记录
if (lastValue != null) {
String historyKey = DEVICE_PREFIX + "data:history:" + deviceId + ":" + dataType;
redisTemplate.executePipelined(
(RedisCallback<Object>)
connection -> {
connection.zAdd(
historyKey.getBytes(), System.currentTimeMillis(), lastValue.getBytes());
connection.zRemRange(historyKey.getBytes(), 0, -101); // 保留最近100条
return null;
});
}
// 更新当前值
redisTemplate.opsForValue().set(currentKey, value, timeout, TimeUnit.valueOf(unit));
} catch (RedisConnectionFailureException e) {
log.error("数据缓存失败 - Device: {}, Type: {}", deviceId, dataType);
} catch (Exception e) {
log.error("数据缓存异常 - Device: {}, Type: {}, Error: {}", deviceId, dataType, e.getMessage());
}
}
// ==================== 分布式锁（设备操作互斥）====================
private static final String LOCK_SCRIPT =
"if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then "
+ " redis.call('pexpire', KEYS[1], ARGV[2]) "
+ " return 1 "
+ "else return 0 end";
private static final String UNLOCK_SCRIPT =
"if redis.call('get', KEYS[1]) == ARGV[1] then "
+ " return redis.call('del', KEYS[1]) "
+ "else return 0 end";
@Comment("获取设备操作锁")
public boolean acquireDeviceLock(String deviceId, String lockId, long expireMs) {
String lockKey = DEVICE_PREFIX + "lock:" + deviceId;
try {
RedisScript<Boolean> script = new DefaultRedisScript<>(LOCK_SCRIPT, Boolean.class);
return Boolean.TRUE.equals(
redisTemplate.execute(
script, Collections.singletonList(lockKey), lockId, String.valueOf(expireMs)));
} catch (RedisConnectionFailureException e) {
log.error("获取锁失败 - Device: {}, LockID: {}", deviceId, lockId);
return false;
} catch (Exception e) {
log.error("获取锁异常 - Device: {}, LockID: {}, Error: {}", deviceId, lockId, e.getMessage());
return false;
}
}
@Comment("释放设备锁")
public boolean releaseDeviceLock(String deviceId, String lockId) {
String lockKey = DEVICE_PREFIX + "lock:" + deviceId;
try {
RedisScript<Long> script = new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class);
Long result = redisTemplate.execute(script, Collections.singletonList(lockKey), lockId);
return result != null && result == 1;
} catch (RedisConnectionFailureException e) {
log.error("释放锁失败 - Device: {}, LockID: {}", deviceId, lockId);
return false;
} catch (Exception e) {
log.error("释放锁异常 - Device: {}, LockID: {}, Error: {}", deviceId, lockId, e.getMessage());
return false;
}
}
// ==================== 辅助方法 ====================
private Map<String, String> convertToStringMap(Map<Object, Object> rawMap) {
Map<String, String> result = new HashMap<>();
rawMap.forEach((k, v) -> result.put(k.toString(), v.toString()));
return result;
}
private void logToLocalStorage(Map<String, String> eventData) {
// 实现本地临时存储逻辑
log.warn("事件降级到本地存储: {}", eventData);
// 实际项目中可写入本地文件或数据库
}
}
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/protocol/magic/ProtocolCodecMagic.java
+38 -11
View File
@@ -142,7 +142,7 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
MagicScript script = MagicScript.create(location, null);
MagicScriptRuntime compile = script.compile();
// 处理内部实现了哪些方法
evalMethodCache(definition, compile);
evalMethodCache(provider, compile);
log.info("编译耗时：{}", (System.currentTimeMillis() - t) + "ms");
t = System.currentTimeMillis();
if (codecMethod.equals(CodecMethod.decode)) {
@@ -159,7 +159,7 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
}
}
private void evalMethodCache(ProtocolSupportDefinition definition, MagicScriptRuntime compile) {
private void evalMethodCache(String provider, MagicScriptRuntime compile) {
String[] varNames = compile.getVarNames();
if (varNames != null && varNames.length > 0) {
Set<String> methods = new HashSet<>();
@@ -170,7 +170,8 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
methods.add(method);
}
}
methodCache.put(definition.getId(), methods);
methodCache.put(provider, methods);
log.info("evalMethodCache provider={}, methods={}", provider, methods);
}
}
@@ -180,6 +181,9 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
if (!magicDecoderProvider.containsKey(decodeRequest.getDefinition().getProvider())) {
// 增加同步锁，防止异常
synchronized (magicDecoderProvider) {
log.info(
"magic decoder not exist,key={}, start reload ",
decodeRequest.getDefinition().getProvider());
load(decodeRequest.getDefinition(), CodecMethod.decode);
}
}
@@ -212,7 +216,12 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
public String encode(ProtocolEncodeRequest encodeRequest) throws CodecException {
try {
if (!magicEncoderProvider.containsKey(encodeRequest.getDefinition().getProvider())) {
load(encodeRequest.getDefinition(), CodecMethod.encode);
log.info(
"magic encode not exist,key={}, start reload ",
encodeRequest.getDefinition().getProvider());
synchronized (magicEncoderProvider) {
load(encodeRequest.getDefinition(), CodecMethod.encode);
}
}
// 如果编解码内部不包含 encode 方法，则直接返回原串
if (methodCache.get(encodeRequest.getDefinition().getId()) == null
@@ -227,7 +236,6 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
context.set("payload", encodeRequest.getPayload());
context.set("context", encodeRequest.getContext());
Object execute = magicScript.execute(context);
log.debug("execute={}", execute.getClass());
return str(execute);
} catch (Exception e) {
String error = ExceptionUtil.getRootCauseMessage(e);
@@ -242,10 +250,18 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
@Override
public void remove(String provider) {
log.info("开始remove magic provider={}", provider);
if (StrUtil.isNotBlank(provider)) {
magicEncoderProvider.remove(provider);
magicDecoderProvider.remove(provider);
magicPreDecoderProvider.remove(provider);
MagicScript encodeV = magicEncoderProvider.remove(provider);
MagicScript decodeV = magicDecoderProvider.remove(provider);
MagicScript preDecodeV = magicPreDecoderProvider.remove(provider);
Set<String> methodV = methodCache.remove(provider);
log.info(
"remove encodeV={},decodeV={},preDecodeV={},methodV={}",
encodeV,
decodeV,
preDecodeV,
methodV);
}
}
@@ -257,7 +273,12 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
try {
if (!magicPreDecoderProvider.containsKey(
protocolDecodeRequest.getDefinition().getProvider())) {
load(protocolDecodeRequest.getDefinition(), CodecMethod.preDecode);
log.info(
"magic preDecode not exist,key={}, start reload ",
protocolDecodeRequest.getDefinition().getProvider());
synchronized (magicPreDecoderProvider) {
load(protocolDecodeRequest.getDefinition(), CodecMethod.preDecode);
}
}
// 如果编解码内部不包含 preDecode 方法，则直接返回原串
if (methodCache.get(protocolDecodeRequest.getDefinition().getId()) == null
@@ -307,7 +328,10 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
public String iotToYour(ProtocolEncodeRequest encodeRequest) throws CodecException {
try {
if (!magicEncoderProvider.containsKey(encodeRequest.getDefinition().getProvider())) {
load(encodeRequest.getDefinition(), CodecMethod.iotToYour);
synchronized (magicEncoderProvider) {
log.info("magic encoder not exist,key={}", encodeRequest.getDefinition().getProvider());
load(encodeRequest.getDefinition(), CodecMethod.iotToYour);
}
}
// 如果编解码内部不包含 iotToYour 方法，则使用encode方法
if (methodCache.get(encodeRequest.getDefinition().getId()) == null
@@ -339,7 +363,10 @@ public class ProtocolCodecMagic extends ProtocolCodecSupportWrapper
public String yourToIot(ProtocolDecodeRequest decodeRequest) throws CodecException {
try {
if (!magicDecoderProvider.containsKey(decodeRequest.getDefinition().getProvider())) {
load(decodeRequest.getDefinition(), CodecMethod.yourToIot);
log.info("magic decoder not exist,key={}", decodeRequest.getDefinition().getProvider());
synchronized (magicDecoderProvider) {
load(decodeRequest.getDefinition(), CodecMethod.yourToIot);
}
}
// 如果编解码内部不包含 yourToIot 方法，则使用decode方法
if (methodCache.get(decodeRequest.getDefinition().getId()) == null
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/AbstractCodecService.java
+8 -2
View File
@@ -119,6 +119,11 @@ public abstract class AbstractCodecService implements ICodecService {
/** 编码 */
@Override
public String encode(String productKey, String payload) {
return encode(productKey, payload, null);
}
@Override
public String encode(String productKey, String payload, Object context) {
ProtocolSupportDefinition protocolDef =
getProtocolDefinitionWithScriptIfNeeded(productKey, CodecMethod.encode);
String result = null;
@@ -127,14 +132,15 @@ public abstract class AbstractCodecService implements ICodecService {
if (protocolDef != null && protocolDef.supportMethod(CodecMethod.encode)) {
ProtocolCodecSupport protocolCodecSupport = getProtocolCodecProvider(protocolDef.getType());
try {
result = protocolCodecSupport.encode(new ProtocolEncodeRequest(protocolDef, payload));
result =
protocolCodecSupport.encode(new ProtocolEncodeRequest(protocolDef, payload, context));
} catch (CodecException e) {
log.error("产品编号={} 原始报文={} 编码报错", productKey, payload, e);
}
}
long t2 = System.currentTimeMillis();
log.info("产品编号={} 原始报文={} 编码={} 耗时={}ms", productKey, payload, result, (t2 - t1));
log.info("产品编号={} 原始报文={} 编码={} 耗时={}ms", productKey, payload, result, (t2 - t1));
return result;
}
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/Codec.java
-48
View File
@@ -1,48 +0,0 @@
/*
*
* Copyright (c) 2025, cn-universal. All Rights Reserved.
*
* @Description: 本文件由 Aleo 开发并拥有版权，未经授权严禁擅自商用、复制或传播。
* @Author: Aleo
* @Email: wo8335224@gmail.com
* @Wechat: outlookFil
*
*
*/
package cn.universal.core.service;
import cn.universal.core.message.UPRequest;
import java.util.List;
public interface Codec {
/**
* 消息解码->物模型
*
* @param productKey 产品ProductKey
* @param payload 实际入参字符串
* @param elementType 格式化类型
* @param <R> 泛型
* @return 返回解码后的结构化消息
*/
<R> List<R> decode(String productKey, String payload, Class<R> elementType);
/**
* 消息解码->物模型
*
* @param productKey 产品ProductKey
* @param payload 实际入参字符串
* @return 返回解码后的结构化消息
*/
default List<UPRequest> decode(String productKey, String payload) {
return null;
}
/**
* 结构化消息->设备识别（二进制、16进制、其他）
*
* @param productKey 产品key
* @param payload 消息的原串
*/
String encode(String productKey, String payload);
}
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/CodecServiceFactory.java
-61
View File
@@ -1,61 +0,0 @@
/*
*
* Copyright (c) 2025, IoT-Universal. All Rights Reserved.
*
* @Description: 本文件由 Aleo 开发并拥有版权，未经授权严禁擅自商用、复制或传播。
* @Author: Aleo
* @Email: wo8335224@gmail.com
* @Wechat: outlookFil
*
*
*/
package cn.universal.core.service;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 编解码服务工厂
*
* <p>用于获取不同协议的编解码服务
*
* @version 1.0 @Author Aleo
* @since 2025/01/20
*/
@Slf4j
@Component
public class CodecServiceFactory {
@Autowired private ICodecService defaultCodecService;
/**
* 获取默认编解码服务
*
* @return 默认编解码服务
*/
public ICodecService getDefaultCodecService() {
return defaultCodecService;
}
/**
* 根据协议类型获取编解码服务
*
* @param protocolType 协议类型
* @return 对应的编解码服务
*/
public ICodecService getCodecService(String protocolType) {
return defaultCodecService;
}
/**
* 检查是否支持指定协议
*
* @param protocolType 协议类型
* @return 是否支持
*/
public boolean isSupported(String protocolType) {
return true; // 默认服务支持所有协议
}
}
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/ICodec.java
-38
View File
@@ -1,38 +0,0 @@
/*
*
* Copyright (c) 2025, IoT-Universal. All Rights Reserved.
*
* @Description: 本文件由 Aleo 开发并拥有版权，未经授权严禁擅自商用、复制或传播。
* @Author: Aleo
* @Email: wo8335224@gmail.com
* @Wechat: outlookFil
*
*
*/
package cn.universal.core.service;
import cn.universal.core.message.UPRequest;
/** 消息编解码 */
public interface ICodec {
default String version() {
return "1.0.0";
}
/** 上行消息消息，预编码 */
default UPRequest preDecode(String productKey, String message) {
return null;
}
/** 进编码前特殊处理 附加影子 */
default String beforeEncode(String productKey, String deviceId, String config, String function) {
return function;
}
/** 下行消息编码 */
default String spliceDown(String productKey, String payload) {
return null;
}
}
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/ICodecService.java
+9
View File
@@ -67,6 +67,15 @@ public interface ICodecService {
*/
String encode(String productKey, String payload);
/**
* 编码
*
* @param productKey 产品Key
* @param payload 原始数据
* @return 编码后的字符串
*/
String encode(String productKey, String payload, Object context);
/**
* 预解码
*
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/IoTDownlFactory.java
+2 -1
View File
@@ -12,6 +12,7 @@
package cn.universal.core.service;
import cn.universal.common.constant.IoTConstant;
import cn.universal.common.domain.R;
import cn.universal.common.exception.IoTException;
import cn.universal.core.protocol.ProtocolModuleInfo;
@@ -109,7 +110,7 @@ public class IoTDownlFactory implements ApplicationContextAware {
switch (operation) {
case "down":
return service.doAction(msg);
case "downPro":
case IoTConstant.DOWN_TO_THIRD_PLATFORM:
return service.downToThirdPlatform(msg);
default:
return R.error("不支持的操作: " + operation);
cn-universal-framework/cn-universal-core/src/main/java/cn/universal/core/service/UniversalCodec.java
-32
View File
@@ -1,32 +0,0 @@
package cn.universal.core.service;
import cn.universal.core.message.UPRequest;
/**
* @version 1.0 @Author Aleo
* @since 2025/6/26 19:41
*/
public interface UniversalCodec extends Codec {
/**
* 预解码，用于TCP表示出哪个ProductKey
*
* @param productKey 产品Key
* @param message 消息原文
* @return 解码内容
*/
default UPRequest preDecode(String productKey, String message) {
return null;
}
/**
* 上行消息转换->转换为阿里云、腾讯云、三方平台
*
* @param productKey 产品Key
* @param message 消息原文
* @return 消息原文
*/
default String messageFormatUP(String productKey, String message) {
return null;
}
}
cn-universal-framework/cn-universal-oss/pom.xml
+5
View File
@@ -32,6 +32,11 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.qiniu</groupId>
<artifactId>qiniu-java-sdk</artifactId>
<version>7.18.0</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
cn-universal-framework/cn-universal-oss/src/main/java/cn/universal/ossm/entity/SysOss.java
+1 -1
View File
@@ -75,6 +75,6 @@ public class SysOss implements Serializable {
private String updateBy;
/** 服务商 */
@Column(name = "third")
@Column(name = "service")
private String service;
}
cn-universal-framework/cn-universal-oss/src/main/java/cn/universal/ossm/oss/enumd/CloudServiceEnumd.java
+2 -1
View File
@@ -13,6 +13,7 @@
package cn.universal.ossm.oss.enumd;
import cn.universal.ossm.oss.service.impl.AliyunCloudStorageServiceImpl;
import cn.universal.ossm.oss.service.impl.QiniuCloudStorageServiceImpl;
import lombok.AllArgsConstructor;
import lombok.Getter;
@@ -22,7 +23,7 @@ import lombok.Getter;
public enum CloudServiceEnumd {
/** 七牛云 */
// QINIU("qiniu", QiniuCloudStorageServiceImpl.class),
QINIU("qiniu", QiniuCloudStorageServiceImpl.class),
/** 阿里云 */
ALIYUN("aliyun", AliyunCloudStorageServiceImpl.class);
cn-universal-framework/cn-universal-oss/src/main/java/cn/universal/ossm/oss/factory/OssFactory.java
+9 -8
View File
@@ -14,6 +14,7 @@ package cn.universal.ossm.oss.factory;
import cn.hutool.core.lang.Assert;
import cn.universal.common.utils.SpringUtils;
import cn.universal.ossm.oss.constant.CloudConstant;
import cn.universal.ossm.oss.enumd.CloudServiceEnumd;
import cn.universal.ossm.oss.service.ICloudStorageService;
import java.util.Map;
@@ -21,18 +22,18 @@ import java.util.concurrent.ConcurrentHashMap;
/** 文件上传Factory @Author Lion Li */
public class OssFactory {
// private static ISysConfigService sysConfigService;
//
// static {
// OssFactory.sysConfigService = SpringUtils.getBean(ISysConfigService.class);
// }
//
// private static ISysConfigService sysConfigService;
//
// static {
// OssFactory.sysConfigService = SpringUtils.getBean(ISysConfigService.class);
// }
private static final Map<String, ICloudStorageService> SERVICES = new ConcurrentHashMap<>();
public static ICloudStorageService instance() {
// String type = sysConfigService.selectConfigByKey(CloudConstant.CLOUD_STORAGE_CONFIG_KEY);
return instance("aliyun");
// String type = sysConfigService.selectConfigByKey(CloudConstant.CLOUD_STORAGE_CONFIG_KEY);
return instance("qiniu");
}
public static ICloudStorageService instance(String type) {
cn-universal-framework/cn-universal-oss/src/main/java/cn/universal/ossm/service/impl/SysOssServiceImpl.java
+9 -6
View File
@@ -13,7 +13,6 @@
package cn.universal.ossm.service.impl;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.universal.common.exception.IoTException;
import cn.universal.ossm.entity.SysOss;
@@ -39,7 +38,7 @@ public class SysOssServiceImpl implements ISysOssService {
@Resource private SysOssMapper sysOssMapper;
@Value("${codec.path:cn-universal}")
@Value("${codec.path:nexiot/}")
private String prePath;
@Override
@@ -58,10 +57,14 @@ public class SysOssServiceImpl implements ISysOssService {
if (file.getOriginalFilename() == null) {
throw new IoTException("文件名为空！");
}
String originalfileName =
prePath + file.getOriginalFilename().replace(".", RandomUtil.randomString(8) + ".");
String suffix =
StrUtil.sub(originalfileName, originalfileName.lastIndexOf("."), originalfileName.length());
// 获取文件扩展名
String originalFilename = file.getOriginalFilename();
String suffix = "";
if (originalFilename != null && originalFilename.contains(".")) {
suffix = StrUtil.sub(originalFilename, originalFilename.lastIndexOf("."), originalFilename.length());
}
// 生成基于 Unix 时间戳的文件名
String originalfileName = prePath + System.currentTimeMillis() + suffix;
ICloudStorageService storage = OssFactory.instance();
UploadResult uploadResult;
try {
cn-universal-framework/pom.xml
+1
View File
@@ -16,6 +16,7 @@
<module>cn-universal-core</module>
<module>cn-universal-exception</module>
<module>cn-universal-config</module>
<module>cn-universal-notice</module>
<module>cn-universal-oss</module>
<module>cn-universal-dm</module>
<!-- <module>cn-universal-rocketmq</module>-->