Merge branch 'dev/0.9_cockpit' into dev/1.0_miniprogram

This commit is contained in:
zhugongping 2026-09-19 04:45:01 +08:00
commit eda5eef6cf
96 changed files with 2693 additions and 293 deletions

View File

@ -72,7 +72,7 @@ services:
logging: logging:
driver: "json-file" driver: "json-file"
options: options:
max-size: "10m" max-size: "50m"
max-file: "2" max-file: "2"
ports: ports:
- 8080:8080 - 8080:8080

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,88 @@
package com.fastbee.web.task;
import com.fastbee.iot.service.IAlertLogService;
import com.fastbee.iot.service.IEventLogService;
import com.fastbee.notify.service.INotifyLogService;
import com.fastbee.quartz.service.ISysJobLogService;
import com.fastbee.system.service.ISysLogininforService;
import com.fastbee.system.service.ISysOperLogService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 日志清理定时任务
*
* @author fastbee
*/
@Component("logCleanTask")
public class LogCleanTask {
private static final Logger log = LoggerFactory.getLogger(LogCleanTask.class);
@Autowired
private IEventLogService eventLogService;
@Autowired
private ISysOperLogService operLogService;
@Autowired
private ISysLogininforService logininforService;
@Autowired
private IAlertLogService alertLogService;
@Autowired
private ISysJobLogService jobLogService;
@Autowired
private INotifyLogService notifyLogService;
// 之后还有清理的表往下写
/**
* 清理日志(默认清理15天前)
*/
public void cleanLogs() {
cleanLogs(15);
}
/**
* 清理日志
*
* @param days 保留天数
*/
public void cleanLogs(int days) {
log.info("========== 开始清理日志数据,保留最近 {} 天的记录 ==========", days);
try {
// 清理 IoT 事件日志
eventLogService.cleanEventLog(days);
//log.info("已清理 iot_event_log 表 {} 天前的数据", days);
// 清理系统操作日志
operLogService.cleanOperLog(days);
//log.info("已清理 sys_oper_log 表 {} 天前的数据", days);
// 清理登录日志
logininforService.cleanLogininfor(days);
//log.info("已清理 sys_logininfor 表 {} 天前的数据", days);
// 清理设备告警日志
alertLogService.cleanAlertLog(days);
//log.info("已清理 iot_alert_log 表 {} 天前的数据", days);
// 清理定时任务日志
jobLogService.cleanJobLog(days);
//log.info("已清理 sys_job_log 表 {} 天前的数据", days);
// 清理通知日志
notifyLogService.cleanNotifyLog(days);
//log.info("已清理 notify_log 表 {} 天前的数据", days);
log.info("========== 日志清理任务执行完成 ==========");
} catch (Exception e) {
log.error("日志清理任务执行失败", e);
throw e;
}
}
}

View File

@ -1,6 +1,7 @@
package com.fastbee.common.config; package com.fastbee.common.config;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.trace.MdcTaskDecorator;
import lombok.Data; import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
@ -60,7 +61,11 @@ public class DeviceTask {
return builder(FastBeeConstant.TASK.FUNCTION_INVOKE_TASK); return builder(FastBeeConstant.TASK.FUNCTION_INVOKE_TASK);
} }
/*内部消费线程*/ /**
* 内部消费线程
* 监听循环从消息恢复 TraceId 并逐条清理;
* TaskDecorator 用于让 @Async 消费方法继承监听线程的 TraceId
*/
@Bean(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK) @Bean(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK)
public Executor messageConsumeTaskExecutor() { public Executor messageConsumeTaskExecutor() {
return builder(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK); return builder(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK);
@ -104,7 +109,7 @@ public class DeviceTask {
} }
/*组装线程池*/ /*组装线程池(带 TaskDecorator,用于跨线程传播 TraceId)*/
private ThreadPoolTaskExecutor builder(String threadNamePrefix) { private ThreadPoolTaskExecutor builder(String threadNamePrefix) {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(coreSize); executor.setCorePoolSize(coreSize);
@ -115,6 +120,10 @@ public class DeviceTask {
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
//线程池名的前缀 //线程池名的前缀
executor.setThreadNamePrefix(threadNamePrefix); executor.setThreadNamePrefix(threadNamePrefix);
// 配置 TaskDecorator,自动传播 TraceId
executor.setTaskDecorator(new MdcTaskDecorator());
executor.initialize(); executor.initialize();
return executor; return executor;
} }

View File

@ -15,6 +15,8 @@ public interface FastBeeConstant {
String MQTT = "mqtt"; String MQTT = "mqtt";
String PORT = "port"; String PORT = "port";
String ADAPTER = "adapter"; String ADAPTER = "adapter";
/** 链路追踪处理器名 */
String TRACE = "trace";
String FRAMEDECODER = "frameDecoder"; String FRAMEDECODER = "frameDecoder";
String DISPATCHER = "dispatcher"; String DISPATCHER = "dispatcher";
String DECODER = "decoder"; String DECODER = "decoder";

View File

@ -8,6 +8,9 @@ import lombok.Data;
@Data @Data
public class DeviceReplyBo { public class DeviceReplyBo {
/** 链路追踪ID */
private String traceId;
/*设备下发消息id*/ /*设备下发消息id*/
private String messageId; private String messageId;
/*标识符*/ /*标识符*/

View File

@ -24,6 +24,9 @@ import java.util.List;
@AllArgsConstructor @AllArgsConstructor
public class DeviceReportBo { public class DeviceReportBo {
/** 链路追踪ID */
private String traceId;
/*设备编号或IMEI号*/ /*设备编号或IMEI号*/
private String serialNumber; private String serialNumber;
/*产品ID*/ /*产品ID*/

View File

@ -18,6 +18,10 @@ import java.util.Date;
@Data @Data
@Builder @Builder
public class DeviceStatusBo { public class DeviceStatusBo {
/** 链路追踪ID */
private String traceId;
/** /**
* 设备客户端id * 设备客户端id
*/ */

View File

@ -11,6 +11,9 @@ import java.util.List;
@Data @Data
public class DeviceTestReportBo { public class DeviceTestReportBo {
/** 链路追踪ID */
private String traceId;
private Long productId; private Long productId;
/** /**
* 设备编号 * 设备编号

View File

@ -17,6 +17,9 @@ import java.util.Map;
@Data @Data
public class InvokeReqDto { public class InvokeReqDto {
/** 链路追踪ID */
private String traceId;
@ApiModelProperty(value = "设备编号") @ApiModelProperty(value = "设备编号")
private String serialNumber; private String serialNumber;

View File

@ -14,6 +14,9 @@ import lombok.NoArgsConstructor;
@NoArgsConstructor @NoArgsConstructor
public class MQSendMessageBo { public class MQSendMessageBo {
/** 链路追踪ID */
private String traceId;
public Boolean isShadow; public Boolean isShadow;
public Long userId; public Long userId;
public Long delay; public Long delay;

View File

@ -17,6 +17,9 @@ import java.math.BigDecimal;
@JsonInclude(JsonInclude.Include.NON_EMPTY) @JsonInclude(JsonInclude.Include.NON_EMPTY)
public class OtaUpgradeBo { public class OtaUpgradeBo {
/** 链路追踪ID */
private String traceId;
// 设备编码 // 设备编码
private String serialNumber; private String serialNumber;
// 升级任务ID // 升级任务ID

View File

@ -17,6 +17,11 @@ import java.util.concurrent.TimeUnit;
@Data @Data
public class OtaUpgradeDelayTask implements Delayed { public class OtaUpgradeDelayTask implements Delayed {
/**
* 链路追踪ID
*/
private String traceId;
/** /**
* 固件id * 固件id
*/ */

View File

@ -0,0 +1,75 @@
package com.fastbee.common.trace;
import com.fastbee.common.trace.TraceContext.TraceSnapshot;
import org.springframework.core.task.TaskDecorator;
import java.util.concurrent.Callable;
/**
* 线程池 TraceId 传播装饰器
* 解决 @Async 异步任务 TraceId 丢失问题
*
* <p>使用方法:
* <pre>
* ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
* executor.setTaskDecorator(new MdcTaskDecorator());
* </pre>
*
* <p>工作原理:
* <ol>
* <li>在任务提交时,捕获当前线程的 TraceContext 快照</li>
* <li>在任务执行时,恢复快照到新线程(MDC 由 TraceContext 统一同步)</li>
* <li>在任务完成后,清理上下文防止线程池污染</li>
* </ol>
*
* <p>说明:MDC 的写入/清理全部由 {@link TraceContext} 统一管理,
* 本装饰器只负责快照与恢复,避免 MDC 与 TraceContext 双重操作导致的不一致。
*
* @author FastBee
*/
public class MdcTaskDecorator implements TaskDecorator {
@Override
public Runnable decorate(Runnable runnable) {
// 捕获当前线程的 TraceContext 快照(跨线程传播的唯一依据)
TraceSnapshot snapshot = TraceContext.snapshot();
return () -> {
if (snapshot != null) {
TraceContext.restore(snapshot);
} else {
// 父线程无上下文,生成新的 TraceId
TraceContext.getOrCreateTraceId();
}
try {
runnable.run();
} finally {
// 清理上下文,防止线程池复用造成污染
TraceContext.clear();
}
};
}
/**
* 装饰 Callable
*
* @param task Callable 任务
* @return 装饰后的 Callable
*/
public <T> Callable<T> decorate(Callable<T> task) {
TraceSnapshot snapshot = TraceContext.snapshot();
return () -> {
if (snapshot != null) {
TraceContext.restore(snapshot);
} else {
TraceContext.getOrCreateTraceId();
}
try {
return task.call();
} finally {
TraceContext.clear();
}
};
}
}

View File

@ -0,0 +1,270 @@
package com.fastbee.common.trace;
import lombok.Data;
import org.slf4j.MDC;
import java.io.Serializable;
/**
* 统一追踪上下文
* 管理 traceId, tenantId, userId 等追踪信息
*
* <p>核心特性:
* <ul>
* <li>基于 ThreadLocal 实现线程隔离</li>
* <li>自动同步到 MDC 用于日志输出</li>
* <li>支持 snapshot/restore 用于跨线程传播</li>
* </ul>
*
* <p>使用示例:
* <pre>
* // Web请求入口
* TraceContext.setTraceId(TraceIdGenerator.generate());
*
* // 异步任务(通过 TaskDecorator 自动传播)
* &#64;Async
* public void asyncTask() {
* log.info("处理中"); // 日志自动包含 traceId
* }
*
* // 手动传播(用于非 Spring 管理的线程池)
* TraceSnapshot snapshot = TraceContext.snapshot();
* executor.submit(() -> {
* TraceContext.restore(snapshot);
* try {
* // 业务逻辑
* } finally {
* TraceContext.clear();
* }
* });
* </pre>
*
* @author FastBee
*/
public class TraceContext {
/**
* TraceId 在 MDC 中的 key
*/
public static final String TRACE_ID_KEY = "traceId";
/**
* TenantId 在 MDC 中的 key
* 预留字段,暂未使用
*/
public static final String TENANT_ID_KEY = "tenantId";
/**
* UserId 在 MDC 中的 key
* 预留字段,暂未使用
*/
public static final String USER_ID_KEY = "userId";
/**
* 线程本地上下文存储
*/
private static final ThreadLocal<TraceSnapshot> CONTEXT = new ThreadLocal<>();
/**
* 追踪上下文快照
* 用于跨线程传递
*/
@Data
public static class TraceSnapshot implements Serializable {
private static final long serialVersionUID = 1L;
private String traceId;
private Long tenantId; // 预留字段
private Long userId; // 预留字段
/**
* 将快照应用到 MDC
* 先移除旧值再写入,保证 MDC 状态与快照精确一致(包括 null 字段)
*/
public void applyToMdc() {
MDC.remove(TRACE_ID_KEY);
MDC.remove(TENANT_ID_KEY);
MDC.remove(USER_ID_KEY);
if (traceId != null) {
MDC.put(TRACE_ID_KEY, traceId);
}
if (tenantId != null) {
MDC.put(TENANT_ID_KEY, String.valueOf(tenantId));
}
if (userId != null) {
MDC.put(USER_ID_KEY, String.valueOf(userId));
}
}
}
// ================== Get Methods ==================
/**
* 获取当前 TraceId
*
* @return TraceId,如果不存在返回 null
*/
public static String getTraceId() {
TraceSnapshot snapshot = CONTEXT.get();
return snapshot != null ? snapshot.getTraceId() : null;
}
/**
* 获取当前 TenantId(预留)
*
* @return TenantId,如果不存在返回 null
*/
public static Long getTenantId() {
TraceSnapshot snapshot = CONTEXT.get();
return snapshot != null ? snapshot.getTenantId() : null;
}
/**
* 获取当前 UserId(预留)
*
* @return UserId,如果不存在返回 null
*/
public static Long getUserId() {
TraceSnapshot snapshot = CONTEXT.get();
return snapshot != null ? snapshot.getUserId() : null;
}
// ================== Set Methods ==================
/**
* 设置 TraceId
* 同时更新 ThreadLocal 和 MDC
*
* @param traceId TraceId
*/
public static void setTraceId(String traceId) {
TraceSnapshot snapshot = CONTEXT.get();
if (snapshot == null) {
snapshot = new TraceSnapshot();
CONTEXT.set(snapshot);
}
snapshot.setTraceId(traceId);
if (traceId != null) {
MDC.put(TRACE_ID_KEY, traceId);
} else {
MDC.remove(TRACE_ID_KEY);
}
}
/**
* 设置 TenantId(预留)
* 同时更新 ThreadLocal 和 MDC
*
* @param tenantId TenantId
*/
public static void setTenantId(Long tenantId) {
TraceSnapshot snapshot = CONTEXT.get();
if (snapshot == null) {
snapshot = new TraceSnapshot();
CONTEXT.set(snapshot);
}
snapshot.setTenantId(tenantId);
if (tenantId != null) {
MDC.put(TENANT_ID_KEY, String.valueOf(tenantId));
} else {
MDC.remove(TENANT_ID_KEY);
}
}
/**
* 设置 UserId(预留)
* 同时更新 ThreadLocal 和 MDC
*
* @param userId UserId
*/
public static void setUserId(Long userId) {
TraceSnapshot snapshot = CONTEXT.get();
if (snapshot == null) {
snapshot = new TraceSnapshot();
CONTEXT.set(snapshot);
}
snapshot.setUserId(userId);
if (userId != null) {
MDC.put(USER_ID_KEY, String.valueOf(userId));
} else {
MDC.remove(USER_ID_KEY);
}
}
// ================== Convenience Methods ==================
/**
* 获取或创建 TraceId
* 如果当前没有 TraceId,自动生成一个新的
*
* @return TraceId
*/
public static String getOrCreateTraceId() {
String traceId = getTraceId();
if (traceId == null || traceId.isEmpty()) {
traceId = TraceIdGenerator.generate();
setTraceId(traceId);
}
return traceId;
}
/**
* 确保当前有 TraceId
* 如果传入的 traceId 不为空,使用传入的;否则生成新的
*
* @param traceId 可选的 TraceId
* @return 最终的 TraceId
*/
public static String ensureTraceId(String traceId) {
if (traceId != null && !traceId.isEmpty()) {
setTraceId(traceId);
return traceId;
}
return getOrCreateTraceId();
}
// ================== Snapshot & Restore ==================
/**
* 创建当前上下文的快照
* 用于跨线程传递
*
* @return TraceSnapshot 快照,如果当前无上下文返回 null
*/
public static TraceSnapshot snapshot() {
TraceSnapshot current = CONTEXT.get();
if (current == null) {
return null;
}
TraceSnapshot copy = new TraceSnapshot();
copy.setTraceId(current.getTraceId());
copy.setTenantId(current.getTenantId());
copy.setUserId(current.getUserId());
return copy;
}
/**
* 从快照恢复上下文
*
* @param snapshot 快照,如果为 null 则清理上下文
*/
public static void restore(TraceSnapshot snapshot) {
if (snapshot == null) {
clear();
return;
}
CONTEXT.set(snapshot);
snapshot.applyToMdc();
}
/**
* 清除当前上下文
* 同时清除 ThreadLocal 和 MDC
*/
public static void clear() {
CONTEXT.remove();
MDC.remove(TRACE_ID_KEY);
MDC.remove(TENANT_ID_KEY);
MDC.remove(USER_ID_KEY);
}
}

View File

@ -0,0 +1,106 @@
package com.fastbee.common.trace;
import org.slf4j.MDC;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
/**
* TraceId 生成器
* 生成纯数字格式的 TraceId
*
* <p>格式:基于时间戳 + 序列号 + 随机数
* <p>示例:752431931275776594
*
* <p>特点:
* <ul>
* <li>纯数字格式,便于存储和索引</li>
* <li>时间有序,大致按生成时间排序</li>
* <li>线程安全,使用 synchronized 保证</li>
* <li>高并发友好,单机每秒可生成百万级 ID</li>
* </ul>
*
* @author FastBee
*/
public class TraceIdGenerator {
/**
* 起始时间戳 (2023-01-01 00:00:00)
* 用于减少时间戳位数,延长 ID 可用年限
*/
private static final long START_TIMESTAMP = 1672531200000L;
/**
* 序列号掩码:16 bit(0 ~ 65535)
*/
private static final long SEQUENCE_MASK = 0xFFFFL;
/**
* 序列号,同一毫秒内递增
*/
private static final AtomicLong sequence = new AtomicLong(0);
/**
* 上次生成时间戳
*/
private static volatile long lastTimestamp = -1L;
/**
* 生成纯数字格式的 TraceId
*
* @return 纯数字 TraceId 字符串
*/
public static String generate() {
return String.valueOf(generateLong());
}
/**
* 生成 long 类型的 TraceId
* 使用 synchronized 保证线程安全
*
* <p>位域分配(低到高):
* <ul>
* <li>随机数 6 bit(第 0~5 位)</li>
* <li>序列号 16 bit(第 6~21 位)</li>
* <li>时间差 剩余位(第 22 位及以上)</li>
* </ul>
*
* @return long 类型 TraceId
*/
public static synchronized long generateLong() {
long timestamp = System.currentTimeMillis();
// 时钟回拨保护:时间不允许倒退,沿用上次时间戳
if (timestamp < lastTimestamp) {
timestamp = lastTimestamp;
}
if (timestamp == lastTimestamp) {
// 同毫秒:序列号递增并回写(& MASK 防止溢出侵入时间位域)
sequence.set((sequence.get() + 1) & SEQUENCE_MASK);
if (sequence.get() == 0) {
// 序列号用尽,等待下一毫秒
timestamp = tilNextMillis(lastTimestamp);
}
} else {
sequence.set(0);
}
lastTimestamp = timestamp;
// 时间差(高位) | 序列号(中位) | 随机数(低位 6 bit)
long timeDiff = timestamp - START_TIMESTAMP;
long random = ThreadLocalRandom.current().nextInt(64);
return (timeDiff << 22) | (sequence.get() << 6) | random;
}
/**
* 等待下一毫秒
*/
private static long tilNextMillis(long lastTimestamp) {
long timestamp = System.currentTimeMillis();
while (timestamp <= lastTimestamp) {
timestamp = System.currentTimeMillis();
}
return timestamp;
}
}

View File

@ -295,21 +295,19 @@ public class StringUtils extends org.apache.commons.lang3.StringUtils {
/** /**
* 判断给定的set列表中是否包含数组array 判断给定的数组array中是否包含给定的元素value * 判断给定的set列表中是否包含数组array 判断给定的数组array中是否包含给定的元素value
* *
* @param set 给定的集合 * @param collection 给定的集合
* @param array 给定的数组 * @param array 给定的数组
* @return boolean 结果 * @return boolean 结果
*/ */
public static boolean containsAny(Collection<String> collection, String... array) { public static boolean containsAny(Collection<String> collection, String... array) {
if (isEmpty(collection) || isEmpty(array)) { if (!isEmpty(collection) && !isEmpty(array)) {
return false;
} else {
for (String str : array) { for (String str : array) {
if (collection.contains(str)) { if (collection.contains(str)) {
return true; return true;
} }
} }
return false;
} }
return false;
} }
/** /**

View File

@ -1,188 +1,64 @@
package com.fastbee.common.utils; package com.fastbee.common.utils;
import com.fastbee.common.utils.uuid.IdUtils; import com.fastbee.common.trace.TraceContext;
import org.slf4j.MDC; import com.fastbee.common.trace.TraceContext.TraceSnapshot;
import java.util.Map;
import java.util.concurrent.Callable; import java.util.concurrent.Callable;
import java.util.concurrent.ThreadLocalRandom;
/** /**
* 日志追踪ID工具类 * 日志追踪ID工具类
* 统一生成TraceId的逻辑,使用IdWorker和ThreadLocalRandom * 提供向后兼容的 API,内部委托给 TraceContext
* *
* @author Sumuxi * @author Sumuxi
* 建议直接使用 {@link TraceContext}
*/ */
public class TraceIdUtils { public class TraceIdUtils {
/** /**
* TraceId在MDC中的key * 包装 Runnable:透传 TraceContext(MDC 由 TraceContext 统一同步)
*/ * 用于非 Spring 管理的线程池(如 ScheduledExecutorService)
public static final String TRACE_ID_KEY = "traceId";
/**
* IdWorker实例(单例模式)
*/
private static volatile IdUtils idWorker;
/**
* 获取IdWorker实例(懒加载单例)
* *
* @return IdWorker实例 * @param task Runnable 任务
*/ * @return 包装后的 Runnable
private static IdUtils getIdWorker() {
if (idWorker == null) {
synchronized (TraceIdUtils.class) {
if (idWorker == null) {
// 使用机器ID,可以根据实际情况配置
// 这里使用1作为默认workerId,可以通过配置文件读取
// IdUtils构造函数:workerId, datacenterId,-1表示随机生成
idWorker = new IdUtils(-1L, -1L);
}
}
}
return idWorker;
}
/**
* 生成TraceId
* 格式:IdWorker.nextId() * 1000 + ThreadLocalRandom.nextInt(1000)
* 参考:generateClueId() 的实现方式
* 使用Math.abs确保TraceId为正数
*
* @return TraceId
*/
public static long generateTraceId() {
long id = getIdWorker().nextId();
// 使用绝对值确保TraceId为正数,避免负数显示
// 注意:Math.abs(id) * 1000 可能会溢出,所以需要对最终结果再次使用Math.abs()
long traceId = Math.abs(id) * 1000 + ThreadLocalRandom.current().nextInt(1000);
// 确保最终结果为非负数(处理溢出情况)
traceId = Math.abs(traceId);
MDC.put(TRACE_ID_KEY, String.valueOf(traceId));
return traceId;
}
/**
* 生成TraceId并放入MDC
* 返回字符串格式的TraceId
*
* @return TraceId字符串
*/
public static String generateTraceIdString() {
long traceId = generateTraceId();
return String.valueOf(traceId);
}
/**
* 设置TraceId到MDC
*
* @param traceId TraceId
*/
public static void setTraceId(String traceId) {
if (traceId != null && !traceId.isEmpty()) {
MDC.put(TRACE_ID_KEY, traceId);
}
}
/**
* 设置TraceId到MDC(long类型)
*
* @param traceId TraceId
*/
public static void setTraceId(long traceId) {
MDC.put(TRACE_ID_KEY, String.valueOf(traceId));
}
/**
* 从MDC获取TraceId
*
* @return TraceId
*/
public static String getTraceId() {
return MDC.get(TRACE_ID_KEY);
}
/**
* 确保当前线程 MDC 中存在 TraceId。
* - 已存在:直接返回
* - 不存在:生成并写入 MDC
*/
public static String ensureTraceId() {
String traceId = getTraceId();
if (traceId == null || traceId.isEmpty()) {
traceId = generateTraceIdString();
}
return traceId;
}
/**
* 使用指定 traceId 覆盖当前线程 MDC(为空则生成)。
* 主要用于跨线程/跨进程透传后的落地。
*/
public static String ensureTraceId(String traceId) {
if (traceId == null || traceId.isEmpty()) {
return ensureTraceId();
}
setTraceId(traceId);
return traceId;
}
/**
* 包装 Runnable:透传提交线程的 MDC,上下文中无 traceId 则补齐;执行完自动清理/还原。
*/ */
public static Runnable wrapWithMdc(Runnable task) { public static Runnable wrapWithMdc(Runnable task) {
final Map<String, String> captured = MDC.getCopyOfContextMap(); final TraceSnapshot snapshot = TraceContext.snapshot();
return () -> { return () -> {
Map<String, String> previous = MDC.getCopyOfContextMap(); if (snapshot != null) {
try { TraceContext.restore(snapshot);
if (captured == null || captured.isEmpty()) {
MDC.clear();
} else { } else {
MDC.setContextMap(captured); TraceContext.getOrCreateTraceId();
} }
ensureTraceId(); try {
task.run(); task.run();
} finally { } finally {
if (previous == null || previous.isEmpty()) { TraceContext.clear();
MDC.clear();
} else {
MDC.setContextMap(previous);
}
} }
}; };
} }
/** /**
* 包装 Callable:透传提交线程的 MDC,上下文中无 traceId 则补齐;执行完自动清理/还原。 * 包装 Callable:透传 TraceContext(MDC 由 TraceContext 统一同步)
* 用于非 Spring 管理的线程池(如 ScheduledExecutorService)
*
* @param task Callable 任务
* @return 包装后的 Callable
*/ */
public static <T> Callable<T> wrapWithMdc(Callable<T> task) { public static <T> Callable<T> wrapWithMdc(Callable<T> task) {
final Map<String, String> captured = MDC.getCopyOfContextMap(); final TraceSnapshot snapshot = TraceContext.snapshot();
return () -> { return () -> {
Map<String, String> previous = MDC.getCopyOfContextMap(); if (snapshot != null) {
try { TraceContext.restore(snapshot);
if (captured == null || captured.isEmpty()) {
MDC.clear();
} else { } else {
MDC.setContextMap(captured); TraceContext.getOrCreateTraceId();
} }
ensureTraceId(); try {
return task.call(); return task.call();
} finally { } finally {
if (previous == null || previous.isEmpty()) { TraceContext.clear();
MDC.clear();
} else {
MDC.setContextMap(previous);
}
} }
}; };
} }
/**
* 清除MDC中的TraceId
*/
public static void clearTraceId() {
MDC.remove(TRACE_ID_KEY);
} }
}

View File

@ -25,12 +25,12 @@ public class CRC8Utils {
*/ */
public static byte calcCrc8_E5(byte[] data) { public static byte calcCrc8_E5(byte[] data) {
byte crc = 0; byte crc = 0;
for (int j = 0; j < data.length; j++) { for (byte datum : data) {
crc ^= data[j]; crc ^= datum;
for (int i = 0; i < 8; i++) { for (int i = 0; i < 8; i++) {
if ((crc & 0x80) != 0) { if ((crc & 0x80) != 0) {
crc = (byte) ((crc) << 1); crc = (byte) ((crc) << 1);
crc ^= 0xE5; crc ^= (byte) 0xE5;
} else { } else {
crc = (byte) ((crc) << 1); crc = (byte) ((crc) << 1);
} }

View File

@ -78,8 +78,7 @@ public class ByteUtils {
byte[] write = new byte[length]; byte[] write = new byte[length];
int index = 0; int index = 0;
for (int i = 0; i < bytes.size(); i++) { for (byte[] values : bytes) {
byte[] values = bytes.get(i);
addBytes(write, values, index); addBytes(write, values, index);
index += values.length; index += values.length;
} }
@ -312,7 +311,8 @@ public class ByteUtils {
if (StringUtils.isBlank(hexStr)) { if (StringUtils.isBlank(hexStr)) {
return null; return null;
} }
if (hexStr.length() % 2 != 0) {//长度为单数 // 长度为单数
if (hexStr.length() % 2 != 0) {
hexStr = "0" + hexStr; // 前面补0 hexStr = "0" + hexStr; // 前面补0
} }

View File

@ -0,0 +1,47 @@
package com.fastbee.framework.aspectj;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.trace.TraceIdGenerator;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
/**
* @Scheduled 定时任务 TraceId 切面
* 自动为所有 Spring 定时任务方法生成 TraceId
*
* @author FastBee
*/
@Slf4j
@Aspect
@Component
@Order(1) // 高优先级,确保最先执行
public class ScheduledTraceAspect {
/**
* 拦截所有 @Scheduled 方法
*/
@Around("@annotation(org.springframework.scheduling.annotation.Scheduled)")
public Object aroundScheduled(ProceedingJoinPoint joinPoint) throws Throwable {
String methodName = joinPoint.getSignature().toShortString();
// 生成 TraceId
String traceId = TraceIdGenerator.generate();
TraceContext.setTraceId(traceId);
log.debug("定时任务开始执行: {}", methodName);
try {
return joinPoint.proceed();
} catch (Throwable e) {
log.error("定时任务执行异常: {}", methodName, e);
throw e;
} finally {
log.debug("定时任务执行完成: {}", methodName);
TraceContext.clear();
}
}
}

View File

@ -1,6 +1,7 @@
package com.fastbee.framework.config; package com.fastbee.framework.config;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.trace.MdcTaskDecorator;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@ -30,6 +31,10 @@ public class OtaThreadPoolConfig {
executor.setThreadNamePrefix(threadNamePrefix); executor.setThreadNamePrefix(threadNamePrefix);
// 线程池对拒绝任务(无线程可用)的处理策略 // 线程池对拒绝任务(无线程可用)的处理策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 新增:配置 TaskDecorator,自动传播 TraceId
executor.setTaskDecorator(new MdcTaskDecorator());
// 进行加载 // 进行加载
executor.initialize(); executor.initialize();
return executor; return executor;

View File

@ -1,5 +1,6 @@
package com.fastbee.framework.config; package com.fastbee.framework.config;
import com.fastbee.common.trace.MdcTaskDecorator;
import com.fastbee.common.utils.Threads; import com.fastbee.common.utils.Threads;
import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
@ -38,6 +39,10 @@ public class ThreadPoolConfig {
executor.setKeepAliveSeconds(keepAliveSeconds); executor.setKeepAliveSeconds(keepAliveSeconds);
// 线程池对拒绝任务(无线程可用)的处理策略 // 线程池对拒绝任务(无线程可用)的处理策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 新增:配置 TaskDecorator,自动传播 TraceId
executor.setTaskDecorator(new MdcTaskDecorator());
return executor; return executor;
} }

View File

@ -1,9 +1,9 @@
package com.fastbee.framework.interceptor; package com.fastbee.framework.interceptor;
import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.trace.TraceIdGenerator;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.web.servlet.HandlerInterceptor; import org.springframework.web.servlet.HandlerInterceptor;
import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse; import javax.servlet.http.HttpServletResponse;
@ -16,24 +16,26 @@ import javax.servlet.http.HttpServletResponse;
@Component @Component
public class TraceIdInterceptor implements HandlerInterceptor { public class TraceIdInterceptor implements HandlerInterceptor {
private static final String TRACE_ID_HEADER = "X-Trace-Id";
@Override @Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
// 获取请求URI // 获取请求URI
String requestURI = request.getRequestURI(); String requestURI = request.getRequestURI();
// 尝试从请求头中获取TraceId(支持分布式追踪) // 尝试从请求头中获取TraceId(支持分布式追踪)
String traceId = request.getHeader("X-Trace-Id"); String traceId = request.getHeader(TRACE_ID_HEADER);
// 如果请求头中没有TraceId,则生成一个新的 // 如果请求头中没有TraceId,则生成一个新的
if (traceId == null || traceId.isEmpty()) { if (traceId == null || traceId.isEmpty()) {
traceId = TraceIdUtils.generateTraceIdString(); traceId = TraceIdGenerator.generate();
} else {
// 如果请求头中有TraceId,设置到MDC
TraceIdUtils.setTraceId(traceId);
} }
// 设置到 TraceContext(自动同步到 MDC)
TraceContext.setTraceId(traceId);
// 将TraceId放入响应头,方便前端获取 // 将TraceId放入响应头,方便前端获取
response.setHeader("X-Trace-Id", traceId); response.setHeader(TRACE_ID_HEADER, traceId);
// 打印请求路径日志 // 打印请求路径日志
org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(TraceIdInterceptor.class); org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(TraceIdInterceptor.class);
@ -44,8 +46,8 @@ public class TraceIdInterceptor implements HandlerInterceptor {
@Override @Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception { public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
// 请求完成后清除MDC中的TraceId,避免内存泄漏 // 请求完成后清除 TraceContext,避免内存泄漏
TraceIdUtils.clearTraceId(); TraceContext.clear();
} }
} }

View File

@ -1,6 +1,7 @@
package com.fastbee.framework.manager; package com.fastbee.framework.manager;
import com.fastbee.common.utils.Threads; import com.fastbee.common.utils.Threads;
import com.fastbee.common.utils.TraceIdUtils;
import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.common.utils.spring.SpringUtils;
import java.util.TimerTask; import java.util.TimerTask;
@ -36,10 +37,17 @@ public class AsyncManager {
/** /**
* 执行任务 * 执行任务
* *
* <p>包装任务以透传 TraceId:{@code scheduledExecutorService} 是裸
* {@link java.util.concurrent.ScheduledThreadPoolExecutor},不支持
* {@code setTaskDecorator},因此必须在提交点包装任务本身。
* {@link TraceIdUtils#wrapWithMdc} 会捕获调用方线程的 TraceContext
* 快照并在 schedule-pool 线程内恢复,使登录/登出/操作日志等
* 异步落库链路的日志携带同一 TraceId。
*
* @param task 任务 * @param task 任务
*/ */
public void execute(TimerTask task) { public void execute(TimerTask task) {
executor.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS); executor.schedule(TraceIdUtils.wrapWithMdc(task), OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS);
} }
/** /**

View File

@ -2,9 +2,10 @@ package com.fastbee.iot.data.job;
import com.fastbee.common.constant.Constants; import com.fastbee.common.constant.Constants;
import com.fastbee.common.constant.ScheduleConstants; import com.fastbee.common.constant.ScheduleConstants;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.trace.TraceIdGenerator;
import com.fastbee.common.utils.ExceptionUtil; import com.fastbee.common.utils.ExceptionUtil;
import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.TraceIdUtils;
import com.fastbee.common.utils.bean.BeanUtils; import com.fastbee.common.utils.bean.BeanUtils;
import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.common.utils.spring.SpringUtils;
import com.fastbee.iot.domain.DeviceJob; import com.fastbee.iot.domain.DeviceJob;
@ -29,7 +30,7 @@ public abstract class DeviceAbstractQuartzJob implements Job {
/** /**
* 线程本地变量 * 线程本地变量
*/ */
private static ThreadLocal<Date> threadLocal = new ThreadLocal<>(); private static final ThreadLocal<Date> threadLocal = new ThreadLocal<>();
@Override @Override
public void execute(JobExecutionContext context) throws JobExecutionException { public void execute(JobExecutionContext context) throws JobExecutionException {
@ -37,9 +38,7 @@ public abstract class DeviceAbstractQuartzJob implements Job {
BeanUtils.copyBeanProp(deviceJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES)); BeanUtils.copyBeanProp(deviceJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES));
try { try {
before(context, deviceJob); before(context, deviceJob);
if (deviceJob != null) {
doExecute(context, deviceJob); doExecute(context, deviceJob);
}
after(context, deviceJob, null); after(context, deviceJob, null);
} catch (Exception e) { } catch (Exception e) {
log.error("任务执行异常 - :", e); log.error("任务执行异常 - :", e);
@ -56,8 +55,8 @@ public abstract class DeviceAbstractQuartzJob implements Job {
protected void before(JobExecutionContext context, DeviceJob deviceJob) { protected void before(JobExecutionContext context, DeviceJob deviceJob) {
threadLocal.set(new Date()); threadLocal.set(new Date());
// 为设备定时任务生成TraceId,使用统一的生成逻辑 // 为设备定时任务生成 TraceId
TraceIdUtils.generateTraceId(); TraceContext.setTraceId(TraceIdGenerator.generate());
log.info("设备定时任务开始执行: {} - {}", deviceJob.getJobName(), deviceJob.getJobGroup()); log.info("设备定时任务开始执行: {} - {}", deviceJob.getJobName(), deviceJob.getJobGroup());
} }
@ -91,10 +90,6 @@ public abstract class DeviceAbstractQuartzJob implements Job {
startTime = new Date(); startTime = new Date();
} }
sysJobLog.setStartTime(startTime); sysJobLog.setStartTime(startTime);
Date stopTime = new Date();
sysJobLog.setStopTime(stopTime);
long runMs = stopTime.getTime() - startTime.getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时:" + runMs + "毫秒");
if (e != null) { if (e != null) {
sysJobLog.setStatus(Constants.FAIL); sysJobLog.setStatus(Constants.FAIL);
String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000); String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000);
@ -103,13 +98,18 @@ public abstract class DeviceAbstractQuartzJob implements Job {
sysJobLog.setStatus(Constants.SUCCESS); sysJobLog.setStatus(Constants.SUCCESS);
} }
Date stopTime = new Date();
sysJobLog.setStopTime(stopTime);
long runMs = stopTime.getTime() - startTime.getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时:" + runMs + "毫秒");
// 写入数据库当中 // 写入数据库当中
SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog); SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog);
// 清除TraceId,避免内存泄漏
TraceIdUtils.clearTraceId();
log.info("设备定时任务执行完成: {} - {}, 耗时: {}ms", deviceJob.getJobName(), deviceJob.getJobGroup(), runMs); log.info("设备定时任务执行完成: {} - {}, 耗时: {}ms", deviceJob.getJobName(), deviceJob.getJobGroup(), runMs);
// 清除 TraceContext
TraceContext.clear();
} }
/** /**

View File

@ -11,6 +11,7 @@ import org.quartz.JobExecutionContext;
*/ */
@DisallowConcurrentExecution @DisallowConcurrentExecution
public class DeviceDisallowConcurrentExecution extends DeviceAbstractQuartzJob { public class DeviceDisallowConcurrentExecution extends DeviceAbstractQuartzJob {
@Override @Override
protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception { protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception {
DeviceJobInvoke.invokeMethod(deviceJob); DeviceJobInvoke.invokeMethod(deviceJob);

View File

@ -9,6 +9,7 @@ import org.quartz.JobExecutionContext;
* @author sumuxi, 2026-04-16 * @author sumuxi, 2026-04-16
*/ */
public class DeviceJobExecution extends DeviceAbstractQuartzJob { public class DeviceJobExecution extends DeviceAbstractQuartzJob {
@Override @Override
protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception { protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception {
DeviceJobInvoke.invokeMethod(deviceJob); DeviceJobInvoke.invokeMethod(deviceJob);

View File

@ -29,13 +29,13 @@ public class DeviceJobInvoke {
/** /**
* 获取消息推送接口 * 获取消息推送接口
*/ */
private static IMqttMessagePublish messagePublish = SpringUtils.getBean(IMqttMessagePublish.class); private static final IMqttMessagePublish messagePublish = SpringUtils.getBean(IMqttMessagePublish.class);
private static SceneMapper sceneMapper = SpringUtils.getBean(SceneMapper.class); private static final SceneMapper sceneMapper = SpringUtils.getBean(SceneMapper.class);
private static FlowLogExecutor flowExecutor = SpringUtils.getBean(FlowLogExecutor.class); private static final FlowLogExecutor flowExecutor = SpringUtils.getBean(FlowLogExecutor.class);
private static IDataHandler dataHandler = SpringUtils.getBean(IDataHandler.class); private static final IDataHandler dataHandler = SpringUtils.getBean(IDataHandler.class);
/** /**
* 执行方法 * 执行方法
@ -44,45 +44,45 @@ public class DeviceJobInvoke {
*/ */
public static void invokeMethod(DeviceJob deviceJob) throws Exception { public static void invokeMethod(DeviceJob deviceJob) throws Exception {
if (deviceJob.getJobType() == 1) { if (deviceJob.getJobType() == 1) {
log.info("------------------------1、执行设备定时任务-----------------------------"); log.info("1、[执行设备定时任务]");
List<Action> actions = JSON.parseArray(deviceJob.getActions(), Action.class); List<Action> actions = JSON.parseArray(deviceJob.getActions(), Action.class);
List<ThingsModelSimpleItem> propertys = new ArrayList<>(); List<ThingsModelSimpleItem> properties = new ArrayList<>();
List<ThingsModelSimpleItem> functions = new ArrayList<>(); List<ThingsModelSimpleItem> functions = new ArrayList<>();
for (int i = 0; i < actions.size(); i++) { for (Action action : actions) {
ThingsModelSimpleItem model = new ThingsModelSimpleItem(); ThingsModelSimpleItem model = new ThingsModelSimpleItem();
model.setId(actions.get(i).getId()); model.setId(action.getId());
model.setValue(actions.get(i).getValue()); model.setValue(action.getValue());
model.setRemark("设备定时"); model.setRemark("设备定时");
if (actions.get(i).getType() == 1) { if (action.getType() == 1) {
propertys.add(model); properties.add(model);
} else if (actions.get(i).getType() == 2) { } else if (action.getType() == 2) {
functions.add(model); functions.add(model);
} }
} }
// 发布属性 // 发布属性
if (propertys.size() > 0) { if (!properties.isEmpty()) {
messagePublish.publishProperty(deviceJob.getProductId(), deviceJob.getSerialNumber(), propertys, 0); messagePublish.publishProperty(deviceJob.getProductId(), deviceJob.getSerialNumber(), properties, 0);
} }
// 发布功能 // 发布功能
if (functions.size() > 0) { if (!functions.isEmpty()) {
messagePublish.publishFunction(deviceJob.getProductId(), deviceJob.getSerialNumber(), functions, 0); messagePublish.publishFunction(deviceJob.getProductId(), deviceJob.getSerialNumber(), functions, 0);
} }
} else if (deviceJob.getJobType() == 3) { } else if (deviceJob.getJobType() == 3) {
log.info("------------------3、[定时执行场景联动]---------------------"); log.info("3、[定时执行场景联动]");
Scene scene = sceneMapper.selectSceneBySceneId(deviceJob.getSceneId()); Scene scene = sceneMapper.selectSceneBySceneId(deviceJob.getSceneId());
// 执行场景规则,异步非阻塞 // 执行场景规则,异步非阻塞
SceneContext context = new SceneContext("", 0L, 0, null); SceneContext context = new SceneContext("", 0L, 0, null);
String requestId = "scene/" + scene.getChainName(); String requestId = "scene/" + scene.getChainName();
flowExecutor.execute2FutureWithRid(scene.getChainName(), null, requestId, context); flowExecutor.execute2FutureWithRid(scene.getChainName(), null, requestId, context);
} else if (4 == deviceJob.getJobType()) { } else if (4 == deviceJob.getJobType()) {
log.info("------------------4、[定时执行场景运算型变量]---------------------"); log.info("4、[定时执行场景运算型变量]");
String s = dataHandler.calculateSceneModelTagValue(deviceJob.getDatasourceId()); String s = dataHandler.calculateSceneModelTagValue(deviceJob.getDatasourceId());
if (StringUtils.isEmpty(s)) { if (StringUtils.isEmpty(s)) {
log.info("------------------[定时执行场景运算型变量失败:+{}]---------------------", s); log.info("[定时执行场景运算型变量失败:+{}]", s);
} }
} else if (5 == deviceJob.getJobType()) { } else if (5 == deviceJob.getJobType()) {
log.info("----------------5、执行modbus轮训指令----------------------"); log.info("5、[执行Modbus轮训指令]");
ModbusPollJob job = new ModbusPollJob(); ModbusPollJob job = new ModbusPollJob();
job.setType(2); job.setType(2);
job.setDeviceJob(deviceJob); job.setDeviceJob(deviceJob);

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceReportBo; import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.DeviceOtherMsgConsumer; import com.fastbee.iot.data.consumer.DeviceOtherMsgConsumer;
import com.fastbee.mq.queue.DeviceOtherQueue; import com.fastbee.mq.queue.DeviceOtherQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -26,7 +27,15 @@ public class DeviceOtherListen {
while (true) { while (true) {
try { try {
DeviceReportBo reportBo = DeviceOtherQueue.take(); DeviceReportBo reportBo = DeviceOtherQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(reportBo.getTraceId());
try {
otherMsgConsumer.consume(reportBo); otherMsgConsumer.consume(reportBo);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("=>emq数据转发异常"); log.error("=>emq数据转发异常");
} }

View File

@ -1,6 +1,7 @@
package com.fastbee.iot.data.listen; package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.DevicePropFetchConsumer; import com.fastbee.iot.data.consumer.DevicePropFetchConsumer;
import com.fastbee.iot.model.modbus.ModbusPollJob; import com.fastbee.iot.model.modbus.ModbusPollJob;
import com.fastbee.mq.queue.DevicePropFetchQueue; import com.fastbee.mq.queue.DevicePropFetchQueue;
@ -27,9 +28,20 @@ public class DevicePropFetchListen {
while (true) { while (true) {
try { try {
ModbusPollJob take = DevicePropFetchQueue.take(); ModbusPollJob take = DevicePropFetchQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(take.getTraceId());
try {
devicePropFetchConsumer.consume(take); devicePropFetchConsumer.consume(take);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("=>设备属性获取异常", e); log.error("=>设备属性获取异常", e);
} finally {
// 确保任何情况下都清理
TraceContext.clear();
} }
} }
} }

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceReportBo; import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.DeviceReplyMsgConsumer; import com.fastbee.iot.data.consumer.DeviceReplyMsgConsumer;
import com.fastbee.mq.queue.DeviceReplyQueue; import com.fastbee.mq.queue.DeviceReplyQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -27,10 +28,21 @@ public class DeviceReplyListen {
try { try {
/*读队列消息*/ /*读队列消息*/
DeviceReportBo reportBo = DeviceReplyQueue.take(); DeviceReportBo reportBo = DeviceReplyQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(reportBo.getTraceId());
try {
/*处理消息*/ /*处理消息*/
deviceReplyMsgHandler.consume(reportBo); deviceReplyMsgHandler.consume(reportBo);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("=>设备回调消息监听异常", e); log.error("=>设备回调消息监听异常", e);
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
} }
} }
} }

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceReportBo; import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.DeviceReportMsgConsumer; import com.fastbee.iot.data.consumer.DeviceReportMsgConsumer;
import com.fastbee.mq.queue.DeviceReportQueue; import com.fastbee.mq.queue.DeviceReportQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -27,10 +28,21 @@ public class DeviceReportListen {
try { try {
/*取出数据*/ /*取出数据*/
DeviceReportBo reportBo = DeviceReportQueue.take(); DeviceReportBo reportBo = DeviceReportQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(reportBo.getTraceId());
try {
/*处理数据*/ /*处理数据*/
reportMsgConsumer.consume(reportBo); reportMsgConsumer.consume(reportBo);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("=>设备上报数据监听异常", e); log.error("=>设备上报数据监听异常", e);
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
} }
} }
} }

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceStatusBo; import com.fastbee.common.core.mq.DeviceStatusBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.DeviceStatusConsumer; import com.fastbee.iot.data.consumer.DeviceStatusConsumer;
import com.fastbee.mq.queue.DeviceStatusQueue; import com.fastbee.mq.queue.DeviceStatusQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -26,10 +27,20 @@ public class DeviceStatusListen {
try { try {
while (true) { while (true) {
DeviceStatusBo status = DeviceStatusQueue.take(); DeviceStatusBo status = DeviceStatusQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(status.getTraceId());
try {
deviceStatusConsumer.consume(status); deviceStatusConsumer.consume(status);
} finally {
TraceContext.clear();
}
} }
} catch (Exception e) { } catch (Exception e) {
log.error("设备状态监听错误", e); log.error("设备状态监听错误", e);
} finally {
TraceContext.clear();
} }
} }

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceTestReportBo; import com.fastbee.common.core.mq.DeviceTestReportBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.DeviceTestConsumer; import com.fastbee.iot.data.consumer.DeviceTestConsumer;
import com.fastbee.mq.queue.DeviceTestQueue; import com.fastbee.mq.queue.DeviceTestQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -26,9 +27,19 @@ public class DeviceTestListen {
try { try {
DeviceTestReportBo take = DeviceTestQueue.take(); DeviceTestReportBo take = DeviceTestQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(take.getTraceId());
try {
deviceTestConsumer.consume(take); deviceTestConsumer.consume(take);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("=>emq数据转发异常"); log.error("=>emq数据转发异常");
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
} }
} }
} }

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.MQSendMessageBo; import com.fastbee.common.core.mq.MQSendMessageBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.consumer.FunctionInvokeConsumer; import com.fastbee.iot.data.consumer.FunctionInvokeConsumer;
import com.fastbee.mq.queue.FunctionInvokeQueue; import com.fastbee.mq.queue.FunctionInvokeQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -26,9 +27,20 @@ public class FunctionInvokeListen {
while (true) { while (true) {
try { try {
MQSendMessageBo sendBo = FunctionInvokeQueue.take(); MQSendMessageBo sendBo = FunctionInvokeQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(sendBo.getTraceId());
try {
functionInvokeConsumer.handler(sendBo); functionInvokeConsumer.handler(sendBo);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("=>下发服务消费异常", e); log.error("=>下发服务消费异常", e);
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
} }
} }
} }

View File

@ -4,6 +4,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.domain.entity.SysMenu; import com.fastbee.common.core.domain.entity.SysMenu;
import com.fastbee.common.core.mq.ota.OtaUpgradeDelayTask; import com.fastbee.common.core.mq.ota.OtaUpgradeDelayTask;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.StringUtils;
import com.fastbee.iot.data.service.IOtaUpgradeService; import com.fastbee.iot.data.service.IOtaUpgradeService;
import com.fastbee.mq.queue.DelayUpgradeQueue; import com.fastbee.mq.queue.DelayUpgradeQueue;
@ -48,12 +49,22 @@ public class UpGradeListener implements ApplicationRunner {
OtaUpgradeDelayTask task = DelayUpgradeQueue.task(); OtaUpgradeDelayTask task = DelayUpgradeQueue.task();
if (StringUtils.isNotNull(task)) { if (StringUtils.isNotNull(task)) {
Date startTime = task.getStartTime(); Date startTime = task.getStartTime();
// 恢复 TraceId
TraceContext.ensureTraceId(task.getTraceId());
try {
otaUpgradeService.upgrade(task); otaUpgradeService.upgrade(task);
log.info("=>开始OTA升级时间{}", startTime); log.info("=>开始OTA升级时间{}", startTime);
} finally {
TraceContext.clear();
}
} }
continue; continue;
} catch (Exception e) { } catch (Exception e) {
log.error("=>OTA升级监听异常", e); log.error("=>OTA升级监听异常", e);
} finally {
TraceContext.clear();
} }
} }
} }

View File

@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.ota.OtaUpgradeBo; import com.fastbee.common.core.mq.ota.OtaUpgradeBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.iot.data.service.IMqttMessagePublish;
import com.fastbee.mq.queue.OtaUpgradeQueue; import com.fastbee.mq.queue.OtaUpgradeQueue;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -28,10 +29,20 @@ public class UpgradeListen {
try { try {
/*获取队列中的OTA升级消息*/ /*获取队列中的OTA升级消息*/
OtaUpgradeBo upgradeBo = OtaUpgradeQueue.take(); OtaUpgradeBo upgradeBo = OtaUpgradeQueue.take();
// 恢复 TraceId
TraceContext.ensureTraceId(upgradeBo.getTraceId());
try {
// OTA升级处理 // OTA升级处理
functionSendService.upGradeOTA(upgradeBo); functionSendService.upGradeOTA(upgradeBo);
} finally {
TraceContext.clear();
}
} catch (Exception e) { } catch (Exception e) {
log.error("->OTA消息监听异常", e); log.error("->OTA消息监听异常", e);
} finally {
TraceContext.clear();
} }
} }
} }

View File

@ -47,6 +47,10 @@ public class DeviceJobServiceImpl implements IDeviceJobService {
@PostConstruct @PostConstruct
public void init() throws SchedulerException, TaskException { public void init() throws SchedulerException, TaskException {
scheduler.clear(); scheduler.clear();
// 打印实际类型
// log.info("Scheduler实际类型: {}", scheduler.getClass().getName());
// log.info("Scheduler元数据: {}", scheduler.getMetaData().getSummary());
// log.info("Scheduler class: {}", scheduler.getClass());
// 设备定时任务 // 设备定时任务
List<DeviceJob> jobList = jobMapper.selectJobAll(); List<DeviceJob> jobList = jobMapper.selectJobAll();
for (DeviceJob deviceJob : jobList) { for (DeviceJob deviceJob : jobList) {

View File

@ -222,7 +222,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
/** /**
* 变量读取 * 变量读取
* *
* @param readVO * @param readVO 前端入参
*/ */
@Override @Override
public void readVariableValue(VariableReadVO readVO) { public void readVariableValue(VariableReadVO readVO) {

View File

@ -6,6 +6,7 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fastbee.common.core.mq.ota.OtaUpgradeDelayTask; import com.fastbee.common.core.mq.ota.OtaUpgradeDelayTask;
import com.fastbee.common.enums.OTAUpgrade; import com.fastbee.common.enums.OTAUpgrade;
import com.fastbee.common.exception.ServiceException; import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.MessageUtils; import com.fastbee.common.utils.MessageUtils;
import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.StringUtils;
@ -141,6 +142,8 @@ public class FirmwareTaskServiceImpl extends ServiceImpl<FirmwareTaskMapper, Fir
throw new ServiceException(MessageUtils.message("firmware.task.upgrade.failed.time.not.valid")); throw new ServiceException(MessageUtils.message("firmware.task.upgrade.failed.time.not.valid"));
} }
// 预定升级的任务放到 延迟任务队列 // 预定升级的任务放到 延迟任务队列
// 记录当前 TraceId,供延迟任务执行时恢复
task.setTraceId(TraceContext.getOrCreateTraceId());
DelayUpgradeQueue.offerTask(task); DelayUpgradeQueue.offerTask(task);
FirmwareTask updatePo = new FirmwareTask(); FirmwareTask updatePo = new FirmwareTask();
updatePo.setId(task.getTaskId()); updatePo.setId(task.getTaskId());

View File

@ -5,6 +5,7 @@ import com.fastbee.common.core.mq.DeviceStatusBo;
import com.fastbee.common.core.mq.DeviceTestReportBo; import com.fastbee.common.core.mq.DeviceTestReportBo;
import com.fastbee.common.core.mq.MQSendMessageBo; import com.fastbee.common.core.mq.MQSendMessageBo;
import com.fastbee.common.core.mq.ota.OtaUpgradeBo; import com.fastbee.common.core.mq.ota.OtaUpgradeBo;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.iot.model.modbus.ModbusPollJob; import com.fastbee.iot.model.modbus.ModbusPollJob;
import com.fastbee.mq.queue.*; import com.fastbee.mq.queue.*;
@ -15,26 +16,55 @@ import com.fastbee.mq.queue.*;
*/ */
public class MessageProducer { public class MessageProducer {
private MessageProducer() {
// 工具类,禁止实例化
}
/*发送设备获取属性消息到队列*/ /*发送设备获取属性消息到队列*/
public static void sendPropFetch(ModbusPollJob deviceJob) { public static void sendPropFetch(ModbusPollJob deviceJob) {
if (deviceJob == null) {
return;
}
// 设置 TraceId
deviceJob.setTraceId(TraceContext.getOrCreateTraceId());
DevicePropFetchQueue.offer(deviceJob); DevicePropFetchQueue.offer(deviceJob);
} }
/*发送设备服务下发消息到队列*/ /*发送设备服务下发消息到队列*/
public static void sendFunctionInvoke(MQSendMessageBo bo) { public static void sendFunctionInvoke(MQSendMessageBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
FunctionInvokeQueue.offer(bo); FunctionInvokeQueue.offer(bo);
} }
/*发送设备上报消息到队列*/ /*发送设备上报消息到队列*/
public static void sendPublishMsg(DeviceReportBo bo) { public static void sendPublishMsg(DeviceReportBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
DeviceReportQueue.offer(bo); DeviceReportQueue.offer(bo);
} }
public static void sendOtherMsg(DeviceReportBo bo) { public static void sendOtherMsg(DeviceReportBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
DeviceOtherQueue.offer(bo); DeviceOtherQueue.offer(bo);
} }
public static void sendStatusMsg(DeviceStatusBo bo) { public static void sendStatusMsg(DeviceStatusBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
DeviceStatusQueue.offer(bo); DeviceStatusQueue.offer(bo);
} }
@ -44,16 +74,31 @@ public class MessageProducer {
* @param bo * @param bo
*/ */
public static void sendDeviceTestMsg(DeviceTestReportBo bo) { public static void sendDeviceTestMsg(DeviceTestReportBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
DeviceTestQueue.offer(bo); DeviceTestQueue.offer(bo);
} }
/*发送OTA消息到队列*/ /*发送OTA消息到队列*/
public static void sendOtaUpgradeMsg(OtaUpgradeBo bo) { public static void sendOtaUpgradeMsg(OtaUpgradeBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
OtaUpgradeQueue.offer(bo); OtaUpgradeQueue.offer(bo);
} }
/*发送设备回复消息到队列*/ /*发送设备回复消息到队列*/
public static void sendDeviceReplyMsg(DeviceReportBo bo) { public static void sendDeviceReplyMsg(DeviceReportBo bo) {
if (bo == null) {
return;
}
// 设置 TraceId
bo.setTraceId(TraceContext.getOrCreateTraceId());
DeviceReplyQueue.offer(bo); DeviceReplyQueue.offer(bo);
} }

View File

@ -58,4 +58,12 @@ public interface NotifyLogMapper {
* @return 结果 * @return 结果
*/ */
public int deleteNotifyLogByIds(Long[] ids); public int deleteNotifyLogByIds(Long[] ids);
/**
* 删除指定天数前的通知日志
*
* @param days 保留天数
* @return 删除记录数
*/
public int deleteNotifyLogBeforeDays(int days);
} }

View File

@ -58,4 +58,11 @@ public interface INotifyLogService {
* @return 结果 * @return 结果
*/ */
public int deleteNotifyLogById(Long id); public int deleteNotifyLogById(Long id);
/**
* 清理通知日志
*
* @param days 保留天数
*/
public void cleanNotifyLog(int days);
} }

View File

@ -132,4 +132,14 @@ public class NotifyLogServiceImpl implements INotifyLogService {
public int deleteNotifyLogById(Long id) { public int deleteNotifyLogById(Long id) {
return notifyLogMapper.deleteNotifyLogById(id); return notifyLogMapper.deleteNotifyLogById(id);
} }
/**
* 清理通知日志
*
* @param days 保留天数
*/
@Override
public void cleanNotifyLog(int days) {
notifyLogMapper.deleteNotifyLogBeforeDays(days);
}
} }

View File

@ -138,4 +138,9 @@
#{id} #{id}
</foreach> </foreach>
</delete> </delete>
<delete id="deleteNotifyLogBeforeDays">
DELETE FROM notify_log
WHERE create_time &lt; DATE_SUB(NOW(), INTERVAL #{days} DAY)
</delete>
</mapper> </mapper>

View File

@ -75,6 +75,13 @@ public class PlatformDeviceMappingController extends BaseController {
// ==================== 子映射管理 ==================== // ==================== 子映射管理 ====================
@PreAuthorize("@ss.hasAnyPermi('platform:deviceMapping:query,platform:deviceMapping:list')")
@GetMapping("/sub-mappings/batch")
@ApiOperation("批量查询多个设备绑定的子设备映射(列表页展开行一次拉取,避免 N 次请求)")
public AjaxResult listSubMappingsBatch(@RequestParam List<Long> parentIds) {
return success(subMappingService.selectByParentIds(parentIds));
}
@PreAuthorize("@ss.hasAnyPermi('platform:deviceMapping:query,platform:deviceMapping:list')") @PreAuthorize("@ss.hasAnyPermi('platform:deviceMapping:query,platform:deviceMapping:list')")
@GetMapping("/{id}/sub-mappings") @GetMapping("/{id}/sub-mappings")
@ApiOperation("查询子设备映射列表") @ApiOperation("查询子设备映射列表")

View File

@ -0,0 +1,96 @@
package com.fastbee.controller.system;
import com.fastbee.common.annotation.Log;
import com.fastbee.common.core.controller.BaseController;
import com.fastbee.common.core.domain.AjaxResult;
import com.fastbee.common.enums.BusinessType;
import com.fastbee.iot.service.IAlertLogService;
import com.fastbee.iot.service.IEventLogService;
import com.fastbee.notify.service.INotifyLogService;
import com.fastbee.quartz.service.ISysJobLogService;
import com.fastbee.system.service.ISysLogininforService;
import com.fastbee.system.service.ISysOperLogService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
* 日志清理Controller
*
* @author fastbee
*/
@Api(tags = "系统管理:日志清理")
@RestController
@RequestMapping("/system/log")
public class LogCleanController extends BaseController {
@Autowired
private IEventLogService eventLogService;
@Autowired
private ISysOperLogService operLogService;
@Autowired
private ISysLogininforService logininforService;
@Autowired
private IAlertLogService alertLogService;
@Autowired
private ISysJobLogService jobLogService;
@Autowired
private INotifyLogService notifyLogService;
/**
* 清理日志数据
*
* @param days 保留天数,默认15天
* @return 清理结果
*/
@ApiOperation("清理日志数据")
@PreAuthorize("@ss.hasRole('admin')")
@Log(title = "日志清理", businessType = BusinessType.CLEAN)
@PostMapping("/clean")
public AjaxResult cleanLogs(
@ApiParam("保留天数,默认15天")
@RequestParam(value = "days", defaultValue = "15") Integer days) {
try {
logger.info("手动触发日志清理,保留最近 {} 天的记录", days);
if (days < 1) {
return error("保留天数必须大于等于1天");
}
// 清理 IoT 事件日志
eventLogService.cleanEventLog(days);
// 清理系统操作日志
operLogService.cleanOperLog(days);
// 清理登录日志
logininforService.cleanLogininfor(days);
// 清理设备告警日志
alertLogService.cleanAlertLog(days);
// 清理定时任务日志
jobLogService.cleanJobLog(days);
// 清理通知日志
notifyLogService.cleanNotifyLog(days);
logger.info("日志清理完成,保留最近 {} 天的记录", days);
return success(String.format("日志清理完成,已删除 %d 天前的历史数据", days));
} catch (Exception e) {
logger.error("日志清理失败", e);
return error("日志清理失败:" + e.getMessage());
}
}
}

View File

@ -254,20 +254,19 @@ public class ToolController extends BaseController {
} }
} catch (Exception ex) { } catch (Exception ex) {
ex.printStackTrace();
log.error("发生错误:{}", ex.getMessage(), ex); log.error("发生错误:{}", ex.getMessage(), ex);
} }
} }
@ApiOperation("mqtt钩子处理") @ApiOperation("mqtt钩子处理")
@PostMapping("/mqtt/webhookv5") @PostMapping("/mqtt/webhookv5")
public void webHookProcessv5(@RequestBody JSONObject json) { public void webHookProcessV5(@RequestBody JSONObject json) {
try { try {
String clientid = json.getString("clientid"); String clientid = json.getString("clientid");
String event = json.getString("event"); String event = json.getString("event");
String peername = json.getString("peername"); // 类似 127.0.0.1:8812 String peername = json.getString("peername"); // 类似 127.0.0.1:8812
String ipAddress = peername; String ipAddress = peername;
if (peername.indexOf(":") != -1) { if (peername.contains(":")) {
ipAddress = peername.substring(0, peername.indexOf(":")); ipAddress = peername.substring(0, peername.indexOf(":"));
} }
log.info("webhook:{},clientid:{},peername:{}", event, clientid, peername); log.info("webhook:{},clientid:{},peername:{}", event, clientid, peername);
@ -310,18 +309,17 @@ public class ToolController extends BaseController {
// 影子模式,发布属性和功能 // 影子模式,发布属性和功能
if (device.getIsShadow() == 1) { if (device.getIsShadow() == 1) {
ThingsModelShadow shadow = deviceService.getDeviceShadowThingsModel(device); ThingsModelShadow shadow = deviceService.getDeviceShadowThingsModel(device);
if (shadow.getProperties().size() > 0) { if (!shadow.getProperties().isEmpty()) {
messagePublish.publishProperty(device.getProductId(), device.getSerialNumber(), shadow.getProperties(), 3); messagePublish.publishProperty(device.getProductId(), device.getSerialNumber(), shadow.getProperties(), 3);
} }
if (shadow.getFunctions().size() > 0) { if (!shadow.getFunctions().isEmpty()) {
messagePublish.publishFunction(device.getProductId(), device.getSerialNumber(), shadow.getFunctions(), 3); messagePublish.publishFunction(device.getProductId(), device.getSerialNumber(), shadow.getFunctions(), 3);
} }
} }
} }
} catch (Exception ex) { } catch (Exception ex) {
ex.printStackTrace(); log.error("发生错误:{}", ex.getMessage());
log.error("发生错误:" + ex.getMessage());
} }
} }

View File

@ -83,9 +83,9 @@ public class SysJobController extends BaseController {
return error(StringUtils.format(MessageUtils.message("job.add.failed.cron.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.add.failed.cron.not.valid"), job.getJobName()));
} else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) { } else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.rmi.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.add.failed.rmi.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS})) { } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.ldap.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.add.failed.ldap.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.HTTP, Constants.HTTPS})) { } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.HTTP, Constants.HTTPS)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.http.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.add.failed.http.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) { } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.string.error"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.add.failed.string.error"), job.getJobName()));
@ -108,9 +108,9 @@ public class SysJobController extends BaseController {
return error(StringUtils.format(MessageUtils.message("job.update.failed.cron.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.update.failed.cron.not.valid"), job.getJobName()));
} else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) { } else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.rmi.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.update.failed.rmi.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS})) { } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.ldap.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.update.failed.ldap.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.HTTP, Constants.HTTPS})) { } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.HTTP, Constants.HTTPS)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.http.not.valid"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.update.failed.http.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) { } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.string.error"), job.getJobName())); return error(StringUtils.format(MessageUtils.message("job.update.failed.string.error"), job.getJobName()));

View File

@ -61,4 +61,12 @@ public interface SysJobLogMapper {
* 清空任务日志 * 清空任务日志
*/ */
public void cleanJobLog(); public void cleanJobLog();
/**
* 删除指定天数前的任务日志
*
* @param days 保留天数
* @return 删除记录数
*/
public int deleteJobLogBeforeDays(int days);
} }

View File

@ -53,4 +53,11 @@ public interface ISysJobLogService {
* 清空任务日志 * 清空任务日志
*/ */
public void cleanJobLog(); public void cleanJobLog();
/**
* 清理指定天数前的任务日志
*
* @param days 保留天数
*/
public void cleanJobLog(int days);
} }

View File

@ -78,4 +78,14 @@ public class SysJobLogServiceImpl implements ISysJobLogService {
public void cleanJobLog() { public void cleanJobLog() {
jobLogMapper.cleanJobLog(); jobLogMapper.cleanJobLog();
} }
/**
* 清理指定天数前的任务日志
*
* @param days 保留天数
*/
@Override
public void cleanJobLog(int days) {
jobLogMapper.deleteJobLogBeforeDays(days);
}
} }

View File

@ -2,9 +2,10 @@ package com.fastbee.quartz.util;
import com.fastbee.common.constant.Constants; import com.fastbee.common.constant.Constants;
import com.fastbee.common.constant.ScheduleConstants; import com.fastbee.common.constant.ScheduleConstants;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.trace.TraceIdGenerator;
import com.fastbee.common.utils.ExceptionUtil; import com.fastbee.common.utils.ExceptionUtil;
import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.TraceIdUtils;
import com.fastbee.common.utils.bean.BeanUtils; import com.fastbee.common.utils.bean.BeanUtils;
import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.common.utils.spring.SpringUtils;
import com.fastbee.quartz.domain.SysJob; import com.fastbee.quartz.domain.SysJob;
@ -29,7 +30,7 @@ public abstract class AbstractQuartzJob implements Job {
/** /**
* 线程本地变量 * 线程本地变量
*/ */
private static ThreadLocal<Date> threadLocal = new ThreadLocal<>(); private static final ThreadLocal<Date> threadLocal = new ThreadLocal<>();
@Override @Override
public void execute(JobExecutionContext context) throws JobExecutionException { public void execute(JobExecutionContext context) throws JobExecutionException {
@ -37,9 +38,7 @@ public abstract class AbstractQuartzJob implements Job {
BeanUtils.copyBeanProp(sysJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES)); BeanUtils.copyBeanProp(sysJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES));
try { try {
before(context, sysJob); before(context, sysJob);
if (sysJob != null) {
doExecute(context, sysJob); doExecute(context, sysJob);
}
after(context, sysJob, null); after(context, sysJob, null);
} catch (Exception e) { } catch (Exception e) {
log.error("任务执行异常 - :", e); log.error("任务执行异常 - :", e);
@ -56,8 +55,8 @@ public abstract class AbstractQuartzJob implements Job {
protected void before(JobExecutionContext context, SysJob sysJob) { protected void before(JobExecutionContext context, SysJob sysJob) {
threadLocal.set(new Date()); threadLocal.set(new Date());
// 为定时任务生成TraceId,使用统一的生成逻辑 // 为定时任务生成 TraceId
TraceIdUtils.generateTraceId(); TraceContext.setTraceId(TraceIdGenerator.generate());
log.info("定时任务开始执行: {} - {}", sysJob.getJobName(), sysJob.getJobGroup()); log.info("定时任务开始执行: {} - {}", sysJob.getJobName(), sysJob.getJobGroup());
} }
@ -77,9 +76,6 @@ public abstract class AbstractQuartzJob implements Job {
sysJobLog.setJobGroup(sysJob.getJobGroup()); sysJobLog.setJobGroup(sysJob.getJobGroup());
sysJobLog.setInvokeTarget(sysJob.getInvokeTarget()); sysJobLog.setInvokeTarget(sysJob.getInvokeTarget());
sysJobLog.setStartTime(startTime); sysJobLog.setStartTime(startTime);
sysJobLog.setStopTime(new Date());
long runMs = sysJobLog.getStopTime().getTime() - sysJobLog.getStartTime().getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时:" + runMs + "毫秒");
if (e != null) { if (e != null) {
sysJobLog.setStatus(Constants.FAIL); sysJobLog.setStatus(Constants.FAIL);
String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000); String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000);
@ -87,14 +83,17 @@ public abstract class AbstractQuartzJob implements Job {
} else { } else {
sysJobLog.setStatus(Constants.SUCCESS); sysJobLog.setStatus(Constants.SUCCESS);
} }
sysJobLog.setStopTime(new Date());
long runMs = sysJobLog.getStopTime().getTime() - sysJobLog.getStartTime().getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时:" + runMs + "毫秒");
// 写入数据库当中 // 写入数据库当中
SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog); SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog);
// 清除TraceId,避免内存泄漏
TraceIdUtils.clearTraceId();
log.info("定时任务执行完成: {} - {}, 耗时: {}ms", sysJob.getJobName(), sysJob.getJobGroup(), runMs); log.info("定时任务执行完成: {} - {}, 耗时: {}ms", sysJob.getJobName(), sysJob.getJobGroup(), runMs);
// 清除 TraceContext
TraceContext.clear();
} }
/** /**

View File

@ -99,8 +99,8 @@ public class JobInvokeUtil {
} }
String[] methodParams = methodStr.split(",(?=([^\"']*[\"'][^\"']*[\"'])*[^\"']*$)"); String[] methodParams = methodStr.split(",(?=([^\"']*[\"'][^\"']*[\"'])*[^\"']*$)");
List<Object[]> classs = new LinkedList<>(); List<Object[]> classs = new LinkedList<>();
for (int i = 0; i < methodParams.length; i++) { for (String methodParam : methodParams) {
String str = StringUtils.trimToEmpty(methodParams[i]); String str = StringUtils.trimToEmpty(methodParam);
// String字符串类型,以'或"开头 // String字符串类型,以'或"开头
if (StringUtils.startsWithAny(str, "'", "\"")) { if (StringUtils.startsWithAny(str, "'", "\"")) {
classs.add(new Object[]{StringUtils.substring(str, 1, str.length() - 1), String.class}); classs.add(new Object[]{StringUtils.substring(str, 1, str.length() - 1), String.class});
@ -151,7 +151,7 @@ public class JobInvokeUtil {
Object[] classs = new Object[methodParams.size()]; Object[] classs = new Object[methodParams.size()];
int index = 0; int index = 0;
for (Object[] os : methodParams) { for (Object[] os : methodParams) {
classs[index] = (Object) os[0]; classs[index] = os[0];
index++; index++;
} }
return classs; return classs;

View File

@ -77,6 +77,11 @@
truncate table sys_job_log truncate table sys_job_log
</update> </update>
<delete id="deleteJobLogBeforeDays">
DELETE FROM sys_job_log
WHERE create_time &lt; DATE_SUB(NOW(), INTERVAL #{days} DAY)
</delete>
<insert id="insertJobLog" parameterType="SysJobLog"> <insert id="insertJobLog" parameterType="SysJobLog">
insert into sys_job_log( insert into sys_job_log(
<if test="jobLogId != null and jobLogId != 0">job_log_id,</if> <if test="jobLogId != null and jobLogId != 0">job_log_id,</if>

View File

@ -47,7 +47,7 @@ public class ModbusRtu extends Message {
protected int start; protected int start;
@Column(length = 1, version = {0x80, 0x81}, desc = "标识位,实例设备是0x80") @Column(length = 1, version = {0x80, 0x81}, desc = "标识位,实例设备是0x80")
protected int mId; protected int mId;
@Column(length = 6, version = 0x80, desc = "MAC地址,6个字节") @Column(length = 6, version = {0x80, 0x81}, desc = "MAC地址,6个字节")
protected String mac; protected String mac;
@Column(length = 1, version = {0x80, 0x81}, desc = "结尾,如:7E") @Column(length = 1, version = {0x80, 0x81}, desc = "结尾,如:7E")
protected int end; protected int end;

View File

@ -39,7 +39,14 @@ public class ModbusEndPoint {
@PakMapping(types = 心跳包) @PakMapping(types = 心跳包)
public void heartbeat(DeviceReport message, Session session) { public void heartbeat(DeviceReport message, Session session) {
//心跳包处理:确保设备注册并更新在线状态
if (!session.isRegistered()) {
session.register(message);
//记录设备信息
DeviceMsg deviceMsg = new DeviceMsg();
deviceMsg.setClientId(message.getClientId());
session.setAttribute(SessionKey.DeviceMsg, deviceMsg);
}
} }
} }

View File

@ -1,5 +1,6 @@
package com.fastbee.record.config; package com.fastbee.record.config;
import com.fastbee.common.trace.MdcTaskDecorator;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableAsync;
@ -52,6 +53,10 @@ public class ThreadPoolTaskConfig {
// 线程池对拒绝任务的处理策略 // 线程池对拒绝任务的处理策略
// CallerRunsPolicy:由调用线程(提交任务的线程)处理该任务 // CallerRunsPolicy:由调用线程(提交任务的线程)处理该任务
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 新增:配置 TaskDecorator,自动传播 TraceId
executor.setTaskDecorator(new MdcTaskDecorator());
// 初始化 // 初始化
executor.initialize(); executor.initialize();
return executor; return executor;

View File

@ -0,0 +1,38 @@
package com.fastbee.base.core.hanler;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.trace.TraceIdGenerator;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* Netty 链路追踪处理器
*
* <p>放置在协议 pipeline 的最前端(编解码器之前),为每条入站报文生成独立的 TraceId,
* 使解码、分发、业务处理全链路的日志都携带同一个 TraceId。
*
* <p>注意:Netty EventLoop 线程会被多连接/多报文复用,
* 因此必须在处理结束后清理,否则会造成 TraceId 串线。
*
* @author FastBee
*/
@ChannelHandler.Sharable
public class TraceHandler extends ChannelInboundHandlerAdapter {
/**
* 共享实例(无状态,可跨 Channel 复用)
*/
public static final TraceHandler INSTANCE = new TraceHandler();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 每条设备报文都是独立业务事件,生成新的 TraceId
TraceContext.setTraceId(TraceIdGenerator.generate());
try {
ctx.fireChannelRead(msg);
} finally {
TraceContext.clear();
}
}
}

View File

@ -1,5 +1,6 @@
package com.fastbee.server; package com.fastbee.server;
import com.fastbee.base.core.hanler.TraceHandler;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.server.config.NettyConfig; import com.fastbee.server.config.NettyConfig;
import com.fastbee.server.handler.*; import com.fastbee.server.handler.*;
@ -61,9 +62,11 @@ public class TCPServer extends Server {
@Override @Override
protected void initChannel(NioSocketChannel channel) throws Exception { protected void initChannel(NioSocketChannel channel) throws Exception {
channel.pipeline() channel.pipeline()
// 第一步、链路追踪(必须位于编解码器之前,让解码日志也带 TraceId)
.addLast(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE)
// 设置心跳时间 // 设置心跳时间
.addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime)) .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime))
// 第一步、添加粘包处理器 // 第二步、添加粘包处理器
.addLast(FastBeeConstant.SERVER.FRAMEDECODER, new ProtocolSniffingDecoder(config.maxFrameLength)) .addLast(FastBeeConstant.SERVER.FRAMEDECODER, new ProtocolSniffingDecoder(config.maxFrameLength))
// 第二步、添加消息适配器 // 第二步、添加消息适配器
.addLast(FastBeeConstant.SERVER.ADAPTER, adapter) .addLast(FastBeeConstant.SERVER.ADAPTER, adapter)

View File

@ -1,5 +1,6 @@
package com.fastbee.server; package com.fastbee.server;
import com.fastbee.base.core.hanler.TraceHandler;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.server.config.NettyConfig; import com.fastbee.server.config.NettyConfig;
import com.fastbee.server.handler.DispatcherHandler; import com.fastbee.server.handler.DispatcherHandler;
@ -52,6 +53,8 @@ public class UDPServer extends Server {
@Override @Override
protected void initChannel(NioDatagramChannel channel) throws Exception { protected void initChannel(NioDatagramChannel channel) throws Exception {
channel.pipeline() channel.pipeline()
// 链路追踪(位于编解码器之前,让解码日志也带 TraceId)
.addLast(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE)
.addLast(FastBeeConstant.SERVER.ADAPTER, adapter) .addLast(FastBeeConstant.SERVER.ADAPTER, adapter)
.addLast(FastBeeConstant.SERVER.DECODER, decoder) .addLast(FastBeeConstant.SERVER.DECODER, decoder)
.addLast(FastBeeConstant.SERVER.ENCODER, encoder) .addLast(FastBeeConstant.SERVER.ENCODER, encoder)

View File

@ -13,6 +13,7 @@ import com.fastbee.common.core.mq.DeviceTestReportBo;
import com.fastbee.common.core.protocol.Message; import com.fastbee.common.core.protocol.Message;
import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem; import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
import com.fastbee.common.enums.ServerType; import com.fastbee.common.enums.ServerType;
import com.fastbee.common.trace.TraceContext;
import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.DateUtils;
import com.fastbee.mq.producer.MessageProducer; import com.fastbee.mq.producer.MessageProducer;
import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandler;
@ -61,6 +62,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter {
s.increment(); s.increment();
} }
/*TraceId 由 pipeline 头部的 TraceHandler 统一生成与清理*/
Packet packet = (Packet) msg; Packet packet = (Packet) msg;
Message request = packet.message; Message request = packet.message;
// 判断是否注册 // 判断是否注册
@ -127,7 +129,16 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter {
} }
} else { } else {
if (handler.async) { if (handler.async) {
executor.execute(() -> channelRead0(ctx, packet, handler)); /*异步执行:手动透传 TraceId(业务线程池非Spring管理,无 TaskDecorator)*/
final String traceId = TraceContext.getTraceId();
executor.execute(() -> {
TraceContext.ensureTraceId(traceId);
try {
channelRead0(ctx, packet, handler);
} finally {
TraceContext.clear();
}
});
} else { } else {
channelRead0(ctx, packet, handler); channelRead0(ctx, packet, handler);
} }

View File

@ -55,6 +55,8 @@ public class MqttMessageDelegate {
log.error("=>客户端未连接"); log.error("=>客户端未连接");
throw new ServiceException("客户端未连接"); throw new ServiceException("客户端未连接");
} }
/*TraceId 由 pipeline 头部的 TraceHandler 统一生成与清理*/
Optional.of(processor.get(messageType)) Optional.of(processor.get(messageType))
.ifPresent(mqttHandler -> mqttHandler.handler(ctx, message)); .ifPresent(mqttHandler -> mqttHandler.handler(ctx, message));
} }

View File

@ -1,5 +1,6 @@
package com.fastbee.mqtt.server; package com.fastbee.mqtt.server;
import com.fastbee.base.core.hanler.TraceHandler;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter; import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter;
import com.fastbee.server.Server; import com.fastbee.server.Server;
@ -54,6 +55,8 @@ public class MqttServer extends Server {
channel.pipeline() channel.pipeline()
.addFirst(FastBeeConstant.SERVER.IDLE .addFirst(FastBeeConstant.SERVER.IDLE
, new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime, TimeUnit.SECONDS)) , new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime, TimeUnit.SECONDS))
// 链路追踪(位于编解码器之前,让解码日志也带 TraceId)
.addFirst(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE)
.addLast(FastBeeConstant.SERVER.DECODER, new MqttDecoder(1024 * 1024 * 2)) .addLast(FastBeeConstant.SERVER.DECODER, new MqttDecoder(1024 * 1024 * 2))
.addLast(FastBeeConstant.SERVER.ENCODER, MqttEncoder.INSTANCE) .addLast(FastBeeConstant.SERVER.ENCODER, MqttEncoder.INSTANCE)
.addLast(messageAdapter); .addLast(messageAdapter);

View File

@ -1,5 +1,6 @@
package com.fastbee.mqtt.server; package com.fastbee.mqtt.server;
import com.fastbee.base.core.hanler.TraceHandler;
import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.mqtt.codec.WebSocketMqttCodec; import com.fastbee.mqtt.codec.WebSocketMqttCodec;
import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter; import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter;
@ -49,6 +50,8 @@ public class WebSocketServer extends Server {
ch.pipeline() ch.pipeline()
.addFirst(FastBeeConstant.WS.HEART_BEAT .addFirst(FastBeeConstant.WS.HEART_BEAT
, new IdleStateHandler(0, 0, 70)) , new IdleStateHandler(0, 0, 70))
/*链路追踪(位于编解码器之前,让解码日志也带 TraceId)*/
.addFirst(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE)
/*http请求响应*/ /*http请求响应*/
.addLast(FastBeeConstant.WS.HTTP_SERVER_CODEC, new HttpServerCodec()) .addLast(FastBeeConstant.WS.HTTP_SERVER_CODEC, new HttpServerCodec())
/*聚合header与body组成完整的Http请求,最大数据量为1Mb*/ /*聚合header与body组成完整的Http请求,最大数据量为1Mb*/

View File

@ -1,5 +1,6 @@
package com.fastbee.sip.conf; package com.fastbee.sip.conf;
import com.fastbee.common.trace.MdcTaskDecorator;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableAsync;
@ -26,6 +27,8 @@ public class ThreadPoolTaskConfig {
executor.setKeepAliveSeconds(keepAliveTime); executor.setKeepAliveSeconds(keepAliveTime);
executor.setThreadNamePrefix(threadNamePrefix); executor.setThreadNamePrefix(threadNamePrefix);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 配置 TaskDecorator,自动传播 TraceId
executor.setTaskDecorator(new MdcTaskDecorator());
executor.initialize(); executor.initialize();
return executor; return executor;
} }

View File

@ -115,4 +115,12 @@ public interface AlertLogMapper {
* @return com.fastbee.iot.model.AlertCountVO * @return com.fastbee.iot.model.AlertCountVO
*/ */
List<AlertCountVO> countAlertLevel(DataCenterParam dataCenterParam); List<AlertCountVO> countAlertLevel(DataCenterParam dataCenterParam);
/**
* 删除指定天数前的告警日志
*
* @param days 保留天数
* @return 删除记录数
*/
int deleteAlertLogBeforeDays(@Param("days") int days);
} }

View File

@ -25,4 +25,12 @@ public interface EventLogMapper extends BaseMapperX<EventLog> {
Long selectEventLogCount(@Param("userId") Long userId); Long selectEventLogCount(@Param("userId") Long userId);
/**
* 删除指定天数前的日志
*
* @param days 保留天数
* @return 删除记录数
*/
int deleteEventLogBeforeDays(@Param("days") int days);
} }

View File

@ -11,6 +11,9 @@ import lombok.Data;
@Data @Data
public class ModbusPollJob { public class ModbusPollJob {
/** 链路追踪ID */
private String traceId;
private ModbusPollMsg pollMsg; private ModbusPollMsg pollMsg;
private DeviceJob deviceJob; private DeviceJob deviceJob;

View File

@ -10,6 +10,9 @@ public interface PlatformDeviceSubMappingMapper extends BaseMapperX<PlatformDevi
List<PlatformDeviceSubMapping> selectByParentId(@Param("parentMappingId") Long parentMappingId); List<PlatformDeviceSubMapping> selectByParentId(@Param("parentMappingId") Long parentMappingId);
/** 批量查询多个父映射下的所有子映射(IN 查询,列表页展开行使用) */
List<PlatformDeviceSubMapping> selectByParentIds(@Param("parentIds") List<Long> parentIds);
PlatformDeviceSubMapping selectByParentAndSerial(@Param("parentMappingId") Long parentMappingId, PlatformDeviceSubMapping selectByParentAndSerial(@Param("parentMappingId") Long parentMappingId,
@Param("subSerialNumber") String subSerialNumber); @Param("subSerialNumber") String subSerialNumber);

View File

@ -4,12 +4,16 @@ import com.baomidou.mybatisplus.extension.service.IService;
import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping;
import java.util.List; import java.util.List;
import java.util.Map;
public interface IPlatformDeviceSubMappingService extends IService<PlatformDeviceSubMapping> { public interface IPlatformDeviceSubMappingService extends IService<PlatformDeviceSubMapping> {
/** 查询父映射下的所有子映射 */ /** 查询父映射下的所有子映射 */
List<PlatformDeviceSubMapping> selectByParentId(Long parentMappingId); List<PlatformDeviceSubMapping> selectByParentId(Long parentMappingId);
/** 批量查询多个父映射下的子映射,按 parentMappingId 分组返回(列表页展开行一次拉取,避免 N 次请求) */
Map<Long, List<PlatformDeviceSubMapping>> selectByParentIds(List<Long> parentIds);
/** 根据子设备序列号查找子映射(推送时使用) */ /** 根据子设备序列号查找子映射(推送时使用) */
List<PlatformDeviceSubMapping> selectBySubSerialNumber(String subSerialNumber); List<PlatformDeviceSubMapping> selectBySubSerialNumber(String subSerialNumber);

View File

@ -3,6 +3,8 @@ package com.fastbee.iot.platform.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.mapper.DeviceMapper;
import com.fastbee.iot.platform.domain.*; import com.fastbee.iot.platform.domain.*;
import com.fastbee.iot.platform.flow.FlowContext; import com.fastbee.iot.platform.flow.FlowContext;
import com.fastbee.iot.platform.flow.PlatformFlowExecutor; import com.fastbee.iot.platform.flow.PlatformFlowExecutor;
@ -21,6 +23,7 @@ import org.springframework.util.StringUtils;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.Objects; import java.util.Objects;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -41,6 +44,8 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl<PlatformDevice
private IPlatformDeviceSubMappingService subMappingService; private IPlatformDeviceSubMappingService subMappingService;
@Autowired @Autowired
private PlatformUserAuthMapper platformUserAuthMapper; private PlatformUserAuthMapper platformUserAuthMapper;
@Autowired
private DeviceMapper deviceMapper;
@Override @Override
public List<PlatformDeviceMapping> selectList(PlatformDeviceMapping query) { public List<PlatformDeviceMapping> selectList(PlatformDeviceMapping query) {
@ -58,7 +63,25 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl<PlatformDevice
.eq(query.getNetStatus() != null, PlatformDeviceMapping::getNetStatus, query.getNetStatus()) .eq(query.getNetStatus() != null, PlatformDeviceMapping::getNetStatus, query.getNetStatus())
.eq(StringUtils.hasText(query.getEnable()), PlatformDeviceMapping::getEnable, query.getEnable()) .eq(StringUtils.hasText(query.getEnable()), PlatformDeviceMapping::getEnable, query.getEnable())
.orderByDesc(PlatformDeviceMapping::getCreateTime); .orderByDesc(PlatformDeviceMapping::getCreateTime);
return this.list(wrapper); List<PlatformDeviceMapping> rows = this.list(wrapper);
// 批量补全设备名称:device_name 为冗余字段,旧数据可能为空,按 deviceId 一次 IN 查询回填,避免前端逐个查询
List<Long> deviceIds = rows.stream()
.filter(m -> m.getDeviceId() != null && !StringUtils.hasText(m.getDeviceName()))
.map(PlatformDeviceMapping::getDeviceId)
.distinct()
.collect(Collectors.toList());
if (!deviceIds.isEmpty()) {
List<Device> devices = deviceMapper.selectList(Wrappers.<Device>lambdaQuery()
.in(Device::getDeviceId, deviceIds));
Map<Long, String> nameMap = devices.stream()
.collect(Collectors.toMap(Device::getDeviceId, Device::getDeviceName, (a, b) -> a));
rows.forEach(m -> {
if (m.getDeviceId() != null && !StringUtils.hasText(m.getDeviceName())) {
m.setDeviceName(nameMap.get(m.getDeviceId()));
}
});
}
return rows;
} }
@Override @Override

View File

@ -8,7 +8,11 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -24,6 +28,25 @@ public class PlatformDeviceSubMappingServiceImpl extends ServiceImpl<PlatformDev
return subMappingMapper.selectByParentId(parentMappingId); return subMappingMapper.selectByParentId(parentMappingId);
} }
@Override
public Map<Long, List<PlatformDeviceSubMapping>> selectByParentIds(List<Long> parentIds) {
if (parentIds == null || parentIds.isEmpty()) {
return Collections.emptyMap();
}
List<PlatformDeviceSubMapping> all = subMappingMapper.selectByParentIds(parentIds);
Map<Long, List<PlatformDeviceSubMapping>> grouped = new HashMap<>();
for (PlatformDeviceSubMapping sub : all) {
Long pid = sub.getParentMappingId();
if (pid == null) continue;
grouped.computeIfAbsent(pid, k -> new ArrayList<>()).add(sub);
}
// 保证请求的每个 parentId 都有 key,即便没有子映射也返回空列表,避免前端取到 undefined
for (Long pid : parentIds) {
grouped.putIfAbsent(pid, new ArrayList<>());
}
return grouped;
}
@Override @Override
public List<PlatformDeviceSubMapping> selectBySubSerialNumber(String subSerialNumber) { public List<PlatformDeviceSubMapping> selectBySubSerialNumber(String subSerialNumber) {
return subMappingMapper.selectBySubSerialNumber(subSerialNumber); return subMappingMapper.selectBySubSerialNumber(subSerialNumber);

View File

@ -107,4 +107,11 @@ public interface IAlertLogService {
* @return com.fastbee.iot.model.AlertCountVO * @return com.fastbee.iot.model.AlertCountVO
*/ */
List<AlertCountVO> countAlertLevel(DataCenterParam dataCenterParam); List<AlertCountVO> countAlertLevel(DataCenterParam dataCenterParam);
/**
* 清理告警日志
*
* @param days 保留天数
*/
void cleanAlertLog(int days);
} }

View File

@ -37,4 +37,11 @@ public interface IEventLogService extends IService<EventLog> {
*/ */
List<HistoryModel> listHistory(EventLog eventLog); List<HistoryModel> listHistory(EventLog eventLog);
/**
* 清理事件日志
*
* @param days 保留天数
*/
void cleanEventLog(int days);
} }

View File

@ -199,4 +199,14 @@ public class AlertLogServiceImpl implements IAlertLogService {
public List<AlertCountVO> countAlertLevel(DataCenterParam dataCenterParam) { public List<AlertCountVO> countAlertLevel(DataCenterParam dataCenterParam) {
return alertLogMapper.countAlertLevel(dataCenterParam); return alertLogMapper.countAlertLevel(dataCenterParam);
} }
/**
* 清理告警日志
*
* @param days 保留天数
*/
@Override
public void cleanAlertLog(int days) {
alertLogMapper.deleteAlertLogBeforeDays(days);
}
} }

View File

@ -210,7 +210,7 @@ public class DataRepairServiceImpl implements IDataRepairService {
public void executeDataRepair(String serialNumber, boolean isShadow, List<DeviceLog> currentLogs) { public void executeDataRepair(String serialNumber, boolean isShadow, List<DeviceLog> currentLogs) {
if (CollectionUtils.isEmpty(currentLogs) || isShadow) { if (CollectionUtils.isEmpty(currentLogs) || isShadow) {
log.debug("无数据或为影子模式,不执行数据修复: serialNumber={}, isShadow={}", serialNumber, isShadow); //log.debug("无数据或为影子模式,不执行数据修复: serialNumber={}, isShadow={}", serialNumber, isShadow);
return; return;
} }

View File

@ -648,8 +648,7 @@ public class DeviceServiceImpl implements IDeviceService {
*/ */
@Override @Override
@DSTransactional @DSTransactional
public List<ThingsModelSimpleItem> reportDeviceThingsModelValue(ThingsModelValuesInput input, int type, public List<ThingsModelSimpleItem> reportDeviceThingsModelValue(ThingsModelValuesInput input, int type, boolean isShadow) {
boolean isShadow) {
String serialNumber = input.getDeviceNumber(); String serialNumber = input.getDeviceNumber();
String key = RedisKeyBuilder.buildTSLVCacheKey(input.getProductId(), serialNumber); String key = RedisKeyBuilder.buildTSLVCacheKey(input.getProductId(), serialNumber);
Map<String, String> maps = new HashMap<>(); Map<String, String> maps = new HashMap<>();
@ -675,6 +674,9 @@ public class DeviceServiceImpl implements IDeviceService {
ThingsModelSimpleItem::getValue, ThingsModelSimpleItem::getValue,
(oldValue, newValue) -> oldValue)); (oldValue, newValue) -> oldValue));
} }
// 查询设备信息
Device device = this.selectDeviceBySerialNumber(serialNumber);
for (ThingsModelSimpleItem item : thingsModelValueRemarkItem) { for (ThingsModelSimpleItem item : thingsModelValueRemarkItem) {
String identity = item.getId(); String identity = item.getId();
if (identity.startsWith("array_")) { if (identity.startsWith("array_")) {
@ -746,11 +748,9 @@ public class DeviceServiceImpl implements IDeviceService {
valueItem.setId(identity); valueItem.setId(identity);
if (id.startsWith("array_")) { if (id.startsWith("array_")) {
// 查询是否有缓存,如果没有先进行缓存 // 查询是否有缓存,如果没有先进行缓存
if (!redisCache.containsKey(key)) { // if (!redisCache.containsKey(key)) {
Device device = this.selectDeviceBySerialNumber(serialNumber); // itslValueCache.addCacheDeviceStatus(input.getProductId(), serialNumber);
this.selectDeviceByDeviceId(device.getDeviceId()); // }
}
int index = Integer.parseInt(id.substring(6, 8)); int index = Integer.parseInt(id.substring(6, 8));
if (isShadow) { if (isShadow) {
String[] shadows = cacheIdentifier.getShadow().split(","); String[] shadows = cacheIdentifier.getShadow().split(",");
@ -791,7 +791,6 @@ public class DeviceServiceImpl implements IDeviceService {
/* ★★★★★★★★★★★★★★★★★★★★★★ 存储数据 - 开始 ★★★★★★★★★★★★★★★★★★★★★★ */ /* ★★★★★★★★★★★★★★★★★★★★★★ 存储数据 - 开始 ★★★★★★★★★★★★★★★★★★★★★★ */
ThingsModelType modelType = ThingsModelType.getType(thingModels.getType()); ThingsModelType modelType = ThingsModelType.getType(thingModels.getType());
Device device = this.selectDeviceBySerialNumber(serialNumber);
switch (modelType) { switch (modelType) {
case PROP: case PROP:
if (1 == thingModels.getIsHistory()) { if (1 == thingModels.getIsHistory()) {

View File

@ -64,4 +64,14 @@ public class EventLogServiceImpl extends ServiceImpl<EventLogMapper, EventLog> i
public List<HistoryModel> listHistory(EventLog eventLog) { public List<HistoryModel> listHistory(EventLog eventLog) {
return eventLogMapper.listHistory(eventLog); return eventLogMapper.listHistory(eventLog);
} }
/**
* 清理事件日志
*
* @param days 保留天数
*/
@Override
public void cleanEventLog(int days) {
eventLogMapper.deleteEventLogBeforeDays(days);
}
} }

View File

@ -274,4 +274,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
</foreach> </foreach>
</delete> </delete>
<delete id="deleteAlertLogBeforeDays">
DELETE FROM iot_alert_log
WHERE create_time &lt; DATE_SUB(NOW(), INTERVAL #{days} DAY)
</delete>
</mapper> </mapper>

View File

@ -59,4 +59,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="userId != null and userId != 0"> and user_id = #{userId}</if> <if test="userId != null and userId != 0"> and user_id = #{userId}</if>
</where> </where>
</select> </select>
<delete id="deleteEventLogBeforeDays">
DELETE FROM iot_event_log
WHERE create_time &lt; DATE_SUB(NOW(), INTERVAL #{days} DAY)
</delete>
</mapper> </mapper>

View File

@ -20,4 +20,14 @@
WHERE sub_serial_number = #{subSerialNumber} WHERE sub_serial_number = #{subSerialNumber}
</select> </select>
<!-- 批量查询多个父映射下的子映射,按 parent_mapping_id 分组,前端列表展开行一次拉取 -->
<select id="selectByParentIds" resultType="com.fastbee.iot.platform.domain.PlatformDeviceSubMapping">
SELECT * FROM platform_device_sub_mapping
WHERE parent_mapping_id IN
<foreach collection="parentIds" item="pid" open="(" separator="," close=")">
#{pid}
</foreach>
ORDER BY parent_mapping_id ASC, id ASC
</select>
</mapper> </mapper>

View File

@ -41,4 +41,12 @@ public interface SysLogininforMapper {
public int cleanLogininfor(); public int cleanLogininfor();
public List<SysLogininfor> getUserCount(); public List<SysLogininfor> getUserCount();
/**
* 删除指定天数前的登录日志
*
* @param days 保留天数
* @return 删除记录数
*/
public int deleteLogininforBeforeDays(int days);
} }

View File

@ -45,4 +45,12 @@ public interface SysOperLogMapper {
* 清空操作日志 * 清空操作日志
*/ */
public void cleanOperLog(); public void cleanOperLog();
/**
* 删除指定天数前的操作日志
*
* @param days 保留天数
* @return 删除记录数
*/
public int deleteOperLogBeforeDays(int days);
} }

View File

@ -42,4 +42,11 @@ public interface ISysLogininforService {
* 统计一周内的登陆数据 * 统计一周内的登陆数据
*/ */
public List<SysLogininfor> getUserCount(); public List<SysLogininfor> getUserCount();
/**
* 清理指定天数前的登录日志
*
* @param days 保留天数
*/
public void cleanLogininfor(int days);
} }

View File

@ -45,4 +45,11 @@ public interface ISysOperLogService {
* 清空操作日志 * 清空操作日志
*/ */
public void cleanOperLog(); public void cleanOperLog();
/**
* 清理指定天数前的操作日志
*
* @param days 保留天数
*/
public void cleanOperLog(int days);
} }

View File

@ -63,4 +63,14 @@ public class SysLogininforServiceImpl implements ISysLogininforService {
public List<SysLogininfor> getUserCount() { public List<SysLogininfor> getUserCount() {
return logininforMapper.getUserCount(); return logininforMapper.getUserCount();
} }
/**
* 清理指定天数前的登录日志
*
* @param days 保留天数
*/
@Override
public void cleanLogininfor(int days) {
logininforMapper.deleteLogininforBeforeDays(days);
}
} }

View File

@ -68,4 +68,14 @@ public class SysOperLogServiceImpl implements ISysOperLogService {
public void cleanOperLog() { public void cleanOperLog() {
operLogMapper.cleanOperLog(); operLogMapper.cleanOperLog();
} }
/**
* 清理指定天数前的操作日志
*
* @param days 保留天数
*/
@Override
public void cleanOperLog(int days) {
operLogMapper.deleteOperLogBeforeDays(days);
}
} }

View File

@ -76,4 +76,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
) b on a.login_time = b.datetime; ) b on a.login_time = b.datetime;
</select> </select>
<delete id="deleteLogininforBeforeDays">
DELETE FROM sys_logininfor
WHERE login_time &lt; DATE_SUB(NOW(), INTERVAL #{days} DAY)
</delete>
</mapper> </mapper>

View File

@ -80,4 +80,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
truncate table sys_oper_log truncate table sys_oper_log
</update> </update>
<delete id="deleteOperLogBeforeDays">
DELETE FROM sys_oper_log
WHERE oper_time &lt; DATE_SUB(NOW(), INTERVAL #{days} DAY)
</delete>
</mapper> </mapper>

View File

@ -0,0 +1,96 @@
-- ============================================
-- 日志清理定时任务初始化 SQL
-- ============================================
-- 1. 插入定时任务记录到 sys_job 表
-- 说明:每天凌晨 0 点自动清理 15 天前的日志数据
-- 注意:LogCleanTask 类位于 fastbee-admin 模块的 com.fastbee.web.task 包下
INSERT INTO `sys_job` (
`job_name`,
`job_group`,
`invoke_target`,
`cron_expression`,
`misfire_policy`,
`concurrent`,
`status`,
`create_by`,
`create_time`,
`remark`
) VALUES (
'日志清理任务', -- 任务名称
'SYSTEM', -- 任务组名
'logCleanTask.cleanLogs(30)', -- 调用目标字符串
'0 0 0 * * ?', -- Cron 表达式:每天凌晨 00:00:00
'3', -- 错误策略:不触发立即执行
'1', -- 禁止并发执行
'0', -- 状态:正常
'admin', -- 创建者
NOW(), -- 创建时间
'每天凌晨0点自动清理15天前的日志数据(iot_event_log、sys_oper_log、sys_logininfor、iot_alert_log、sys_job_log、notify_log)'
);
-- ============================================
-- 清理的日志表说明
-- ============================================
-- 1. iot_event_log - IoT 事件日志
-- 2. sys_oper_log - 系统操作日志
-- 3. sys_logininfor - 登录日志
-- 4. iot_alert_log - 设备告警日志
-- 5. sys_job_log - 定时任务日志
-- 6. notify_log - 通知日志
--
-- 注意:iot_device_log 表(设备监测日志)不参与清理
-- 该表为核心数据,目前有 214 万条记录,需要数据迁移方案
-- 2. 可选:添加日志清理权限菜单(根据需要执行)
-- 如果需要在管理后台显示权限按钮,可以执行以下 SQL
/*
INSERT INTO `sys_menu` (
`menu_name`,
`parent_id`,
`order_num`,
`path`,
`component`,
`query`,
`is_frame`,
`is_cache`,
`menu_type`,
`visible`,
`status`,
`perms`,
`icon`,
`create_by`,
`create_time`,
`remark`
) VALUES (
'日志清理', -- 菜单名称
2, -- 父菜单ID(根据实际情况调整,2通常是系统管理)
99, -- 显示顺序
'', -- 路由地址
'', -- 组件路径
'', -- 路由参数
1, -- 不是外链
0, -- 不缓存
'F', -- 菜单类型:F=按钮
'0', -- 显示状态
'0', -- 菜单状态
'system:log:clean', -- 权限标识
'#', -- 图标
'admin', -- 创建者
NOW(), -- 创建时间
'日志清理按钮权限'
);
*/
-- ============================================
-- 验证定时任务是否创建成功
-- ============================================
-- SELECT * FROM sys_job WHERE job_name = '日志清理任务';
-- ============================================
-- 手动触发清理(可选)
-- ============================================
-- 如果需要立即清理历史数据,可以通过以下方式:
-- 方式1:通过管理后台 -> 系统监控 -> 定时任务 -> 点击"执行一次"
-- 方式2:调用 API 接口:POST /system/log/clean?days=30
-- 方式3:数据库执行:UPDATE sys_job SET status = '0' WHERE job_name = '日志清理任务';

View File

@ -28,6 +28,11 @@ export function listSubMappings(parentId) {
return request({ url: `/platform/device-mapping/${parentId}/sub-mappings`, method: 'get' }) return request({ url: `/platform/device-mapping/${parentId}/sub-mappings`, method: 'get' })
} }
// 批量查询多个设备绑定的子设备映射,列表页展开行一次拉取,避免 N 次请求
export function listSubMappingsBatch(parentIds) {
return request({ url: '/platform/device-mapping/sub-mappings/batch', method: 'get', params: { parentIds: parentIds.join(',') } })
}
export function syncSubMappings(parentId, data) { export function syncSubMappings(parentId, data) {
return request({ url: `/platform/device-mapping/${parentId}/sub-mappings`, method: 'post', data }) return request({ url: `/platform/device-mapping/${parentId}/sub-mappings`, method: 'post', data })
} }

View File

@ -335,7 +335,7 @@
</template> </template>
<script> <script>
import { listPlatformDeviceMapping, addPlatformDeviceMapping, updatePlatformDeviceMapping, delPlatformDeviceMapping, onboardDevice, listSubMappings, syncSubMappings } from '@/api/platform/deviceMapping' import { listPlatformDeviceMapping, addPlatformDeviceMapping, updatePlatformDeviceMapping, delPlatformDeviceMapping, onboardDevice, listSubMappings, listSubMappingsBatch, syncSubMappings } from '@/api/platform/deviceMapping'
import { listPlatformInstance } from '@/api/platform/instance' import { listPlatformInstance } from '@/api/platform/instance'
import { getInitializedUsers, getProjectsByUserAuthId } from '@/api/platform/userAuth' import { getInitializedUsers, getProjectsByUserAuthId } from '@/api/platform/userAuth'
import { listPlatformInterface } from '@/api/platform/interface' import { listPlatformInterface } from '@/api/platform/interface'
@ -425,25 +425,15 @@ export default {
this.loading = true this.loading = true
listPlatformDeviceMapping(this.queryParams).then(res => { listPlatformDeviceMapping(this.queryParams).then(res => {
this.list = res.rows; this.total = res.total; this.loading = false this.list = res.rows; this.total = res.total; this.loading = false
// 加载设备名称映射 // deviceName 由后端列表查询批量补全,前端无需再逐个 getDevice(N→0)
const ids = [...new Set(res.rows.map(r => r.deviceId).filter(Boolean))] // 预加载网关设备的子映射(批量一次拉取,避免 N 次请求)
if (ids.length) { this.subMappingCache = {}
Promise.all(ids.map(id => getDevice(id).catch(() => null))).then(results => { const gatewayIds = res.rows.filter(r => r.deviceType === 1).map(r => r.id).filter(Boolean)
const map = {} if (gatewayIds.length) {
results.forEach(res => { listSubMappingsBatch(gatewayIds).then(subRes => {
const d = res && (res.data || res) this.subMappingCache = subRes.data || {}
if (d && d.deviceId) map[d.deviceId] = d.deviceName
})
this.deviceMap = map
}) })
} }
// 预加载网关设备的子映射
this.subMappingCache = {}
res.rows.filter(r => r.deviceType === 1).forEach(row => {
listSubMappings(row.id).then(subRes => {
this.$set(this.subMappingCache, row.id, subRes.data || [])
})
})
}) })
}, },
platformName(id) { const p = this.platformList.find(x => x.id === id); return p ? p.name : id }, platformName(id) { const p = this.platformList.find(x => x.id === id); return p ? p.name : id },