diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index fa154571..ed2c0423 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -72,7 +72,7 @@ services: logging: driver: "json-file" options: - max-size: "10m" + max-size: "50m" max-file: "2" ports: - 8080:8080 diff --git a/docs/knowledge-base/07-TraceId 链路追踪实现.md b/docs/knowledge-base/07-TraceId 链路追踪实现.md new file mode 100644 index 00000000..56095386 --- /dev/null +++ b/docs/knowledge-base/07-TraceId 链路追踪实现.md @@ -0,0 +1,1213 @@ +# 任务：TraceId 链路追踪与上下文传播能力 + +你现在需要为我的 Java 单体项目实现一套“统一 TraceId 链路追踪基础设施”。 + +## 一、项目背景 + +当前项目是 Java 单体应用，技术栈： + +- JDK 8 +- Spring Boot 2.5.x +- Spring MVC +- Logback / SLF4J +- MySQL +- Redis +- Quartz +- MQ +- Netty +- MQTT +- 后续会接入 Python AI 服务、LLM、微服务、OpenTelemetry + +项目未来会逐步发展为： + +```text +单体应用 + ↓ +IoT + SaaS + AI + ↓ +微服务 + ↓ +OpenTelemetry / 分布式链路追踪 +``` + +因此这次实现不能只是简单地增加一个 MDC 字段，而要设计成可长期演进的基础框架。 + +--- + +# 二、核心目标 + +实现统一的： + +```text +TraceContext +``` + +核心字段至少包括： + +```text +traceId +tenantId +userId +``` + +其中当前阶段最重要的是： + +```text +traceId +``` + +核心原则： + +> 一次完整业务执行链使用同一个 traceId。 + +例如： + +```text +前端请求 + ↓ +Controller + ↓ +Service + ↓ +线程池 + ↓ +MQ + ↓ +MQ Consumer + ↓ +AI 服务 + ↓ +数据库 + ↓ +通知 +``` + +如果它们属于同一次业务执行，则全部使用： + +```text +traceId = A +``` + +--- + +# 三、TraceId 生命周期规则 + +必须严格遵循以下规则。 + +## 1. 有上游 TraceId + +优先继承。 + +例如 HTTP： + +```http +X-Trace-Id: 01KABC... +``` + +后端收到后： + +```text +TraceId = 01KABC... +``` + +不要重新生成。 + +--- + +## 2. 没有上游 TraceId + +自动创建新的 TraceId。 + +例如： + +```text +浏览器 + ↓ +HTTP + ↓ +Java +``` + +如果 HTTP 没有： + +```http +X-Trace-Id +``` + +则 Java 自动生成。 + +--- + +## 3. 定时任务没有上游 TraceId + +Quartz / @Scheduled 等任务启动时自动生成： + +```text +traceId +``` + +例如： + +```text +traceId=T202609150001 +``` + +整个任务执行过程保持相同 TraceId。 + +--- + +# 四、TraceId 生成器 + +设计独立： + +```java +TraceIdGenerator +``` + +不要把 TraceId 生成逻辑写死在 Filter 中。 + +要求： + +```text +TraceContext +TraceIdGenerator +``` + +职责分离。 + +优先考虑 ULID 或项目中已有的高质量 ID 生成方案。 + +要求： + +- 唯一 +- 高并发安全 +- 尽量时间有序 +- 适合日志检索 +- 不依赖数据库 +- 后续可以替换实现 + +--- + +# 五、TraceContext 设计 + +创建统一： + +```java +TraceContext +``` + +例如： + +```text +TraceContext +├── getTraceId() +├── setTraceId() +├── getOrCreateTraceId() +├── set() +├── get() +├── clear() +└── snapshot() +``` + +不要让业务代码直接到处： + +```java +MDC.put("traceId", ...) +``` + +业务代码应该依赖： + +```java +TraceContext +``` + +MDC 只是 TraceContext 的一种底层日志传播实现。 + +核心原则： + +```text +TraceContext ≠ MDC +``` + +而应该是： + +```text +TraceContext + ↓ +MDC +HTTP Header +MQ Header +ThreadPool +Future +``` + +--- + +# 六、Web 请求 + +实现 Spring MVC Filter。 + +建议： + +```java +OncePerRequestFilter +``` + +功能： + +1. 从 HTTP Header 获取： + +```http +X-Trace-Id +``` + +2. 如果存在且合法： + +```text +继续使用 +``` + +3. 如果不存在： + +```text +生成 TraceId +``` + +4. 放入： + +```text +TraceContext +MDC +``` + +5. 将 TraceId 写入 HTTP Response Header： + +```http +X-Trace-Id: 01KABC... +``` + +6. 请求结束后必须清理： + +```text +MDC +TraceContext +``` + +防止 Tomcat 线程池复用造成 TraceId 污染。 + +--- + +# 七、日志输出 + +修改 Logback 日志格式，使所有日志自动包含： + +```text +traceId +``` + +例如： + +```text +2026-09-15 11:20:31.123 INFO +[http-nio-8080-exec-5] +traceId=01KABC123 +DeviceService +- +查询设备成功 deviceId=DEV001 +``` + +要求： + +- 控制台日志包含 traceId +- 普通文件日志包含 traceId +- error 日志包含 traceId +- 不要求业务代码修改每一条 log.info() +- 使用 MDC 自动完成 + +同时考虑 JSON 日志扩展能力。 + +--- + +# 八、线程池上下文传播 + +这是本任务的重点。 + +普通： + +```java +ThreadPoolExecutor +``` + +或者： + +```java +CompletableFuture +``` + +不能保证 MDC 自动传播。 + +必须实现统一的上下文传播机制。 + +例如： + +```text +TraceContext + ↓ +snapshot + ↓ +Runnable / Callable + ↓ +restore + ↓ +执行 + ↓ +clear +``` + +要求实现类似： + +```text +TraceContextRunnable +TraceContextCallable +TraceableExecutor +``` + +或者使用 Spring 提供的： + +```text +TaskDecorator +``` + +如果项目使用 Spring ThreadPoolTaskExecutor，优先通过： + +```text +TaskDecorator +``` + +实现。 + +目标： + +```text +Web线程 +traceId=A + ↓ +线程池 +traceId=A +``` + +例如： + +```java +executor.execute(() -> { + log.info("异步处理"); +}); +``` + +日志必须仍然是： + +```text +traceId=A +``` + +--- + +# 九、CompletableFuture + +重点处理： + +```java +CompletableFuture.runAsync(...) +CompletableFuture.supplyAsync(...) +thenApply(...) +thenAccept(...) +thenRun(...) +``` + +分析项目中现有异步代码。 + +不要简单粗暴地修改所有业务代码。 + +优先通过统一线程池： + +```text +TraceableExecutor +``` + +解决。 + +如果无法完全覆盖 CompletableFuture 的默认线程池，需要明确指出风险，并提供统一解决方案。 + +禁止让业务代码到处手动： + +```java +MDC.put(...) +MDC.clear(...) +``` + +--- + +# 十、MQ TraceId 传播 + +这是第二个重点。 + +当 Java 发送 MQ 消息时： + +必须把当前： + +```text +traceId +``` + +写入 MQ Message Header / Properties。 + +例如： + +```text +traceId=A +``` + +发送： + +```text +MQ Message +├── Header +│ └── traceId=A +└── Body +``` + +Consumer 收到消息： + +```text +MQ Header + ↓ +读取 traceId + ↓ +TraceContext.setTraceId() + ↓ +MDC + ↓ +业务处理 +``` + +这样： + +```text +Web +traceId=A + ↓ +MQ Producer +traceId=A + ↓ +MQ +traceId=A + ↓ +Consumer +traceId=A +``` + +如果 MQ 消息没有 TraceId： + +```text +Consumer + ↓ +自动生成新的 TraceId +``` + +但正常情况下必须继承 Producer 的 TraceId。 + +--- + +# 十一、MQ 不同消息中间件 + +先检查当前项目实际使用的 MQ。 + +不要凭空假设。 + +如果项目目前只有一种 MQ： + +优先实现当前实际使用的 MQ。 + +同时将设计抽象成： + +```text +TraceMessageInterceptor +``` + +未来可以支持： + +```text +RabbitMQ +Kafka +RocketMQ +Redis Stream +其他 MQ +``` + +不要为了扩展性一次性实现大量没有使用的 MQ。 + +--- + +# 十二、HTTP 调用链 + +如果 Java 单体项目会调用其他服务： + +例如： + +```text +Java + ↓ HTTP +Python AI +``` + +必须自动传递： + +```http +X-Trace-Id: A +``` + +如果项目使用： + +```text +RestTemplate +WebClient +OpenFeign +HttpClient +OkHttp +``` + +检查实际使用情况。 + +优先通过统一 Interceptor 实现。 + +例如： + +```text +Java +traceId=A + ↓ +HTTP Header +X-Trace-Id=A + ↓ +Python AI +traceId=A +``` + +这样 Java 和 Python 日志可以使用同一个 TraceId 检索。 + +--- + +# 十三、Netty / MQTT / TCP / UDP + +这是 IoT 项目的重要场景。 + +检查项目现有： + +```text +Netty +MQTT +TCP +UDP +CoAP +``` + +入口。 + +对于每一个独立的消息 / 请求： + +如果存在上游 TraceId： + +```text +继承 +``` + +否则： + +```text +生成 +``` + +然后进入： + +```text +TraceContext +``` + +例如： + +```text +设备 DEV001 + ↓ +MQTT + ↓ +traceId=M001 + ↓ +数据解析 + ↓ +物模型 + ↓ +数据库 + ↓ +告警 +``` + +整条处理链使用： + +```text +traceId=M001 +``` + +注意： + +> 设备每次独立上报消息属于独立业务事件，因此通常应该生成新的 TraceId，而不是让一个设备永久使用同一个 TraceId。 + +--- + +# 十五、SaaS 上下文 + +项目未来是 SaaS，因此 TraceContext 预留： + +```text +traceId +tenantId +userId +``` + +如果当前项目已经有： + +```text +tenantId +userId +``` + +则在请求认证成功后自动放入上下文。 + +例如： + +```text +traceId=A +tenantId=10001 +userId=9527 +``` + +但不要因为本次任务而大范围修改现有权限系统。 + +只需要提供统一上下文承载能力。 + +--- + +# 十六、线程安全 + +这是重点。 + +必须考虑： + +```text +Tomcat线程池复用 +Quartz线程池 +业务线程池 +CompletableFuture +MQ Consumer线程 +Netty EventLoop +``` + +任何任务结束后都必须： + +```text +clear +``` + +禁止出现： + +```text +请求A +traceId=A + +请求结束 + +请求B +traceId=A +``` + +这种严重的数据串链问题。 + +--- + +# 十七、异常处理 + +异常情况下 TraceId 也必须保留。 + +例如： + +```text +traceId=A +ERROR +设备数据处理失败 +``` + +完整异常堆栈必须能够通过： + +```text +traceId=A +``` + +检索出来。 + +Web 异常响应最好也返回： + +```http +X-Trace-Id: A +``` + +这样前端报错： + +```text +TraceId: A +``` + +我就可以直接去 Kibana / 日志文件搜索： + +```text +traceId=A +``` + +定位问题。 + +--- + +# 十八、日志检索目标 + +最终必须支持： + +```bash +grep "traceId=01KABC" app.log +``` + +或者 Kibana： + +```text +traceId:"01KABC" +``` + +一次找到： + +```text +HTTP请求 +Service +数据库 +线程池 +MQ Producer +MQ Consumer +AI +异常 +``` + +--- + +# 十九、代码结构要求 + +根据当前项目实际结构，合理放置。 + +如果存在类似： + +```text +common +framework +core +starter +``` + +优先放到公共基础模块。 + +推荐结构： + +```text +xxx-core +└── trace + ├── TraceContext + ├── TraceIdGenerator + ├── TraceContextSnapshot + ├── TraceContextRunnable + └── TraceContextCallable + +xxx-web-starter +└── trace + └── TraceIdFilter + +xxx-log-starter +└── trace + └── MDC / Logback + +xxx-schedule-starter +└── trace + └── Quartz Trace + +xxx-mq-starter +└── trace + ├── Producer Trace + └── Consumer Trace +``` + +具体目录必须根据当前项目实际结构调整，不要机械套用。 + +--- + +# 二十、不要过度设计 + +当前项目是单体项目。 + +第一阶段： + +必须实现： + +```text +TraceId +MDC +Web +ThreadPool +Quartz +MQ +HTTP +``` + +暂时不要自行实现完整： + +```text +Span +Trace Graph +Jaeger +Zipkin +SkyWalking +OpenTelemetry Collector +``` + +但是代码设计要为未来 OpenTelemetry 留出扩展点。 + +也就是说： + +```text +现在： +TraceId + MDC + +未来： +TraceId + SpanId + ParentSpanId + OpenTelemetry +``` + +不要把当前 TraceContext 设计成未来无法迁移的死结构。 + +--- + +# 二十一、需要先做代码分析 + +在开始修改代码之前： + +1. 扫描整个项目结构。 +2. 找到 Spring Boot 启动模块。 +3. 找到 Web 请求入口。 +4. 找到现有 Filter / Interceptor。 +5. 找到现有线程池。 +6. 找到 CompletableFuture。 +7. 找到 Quartz / @Scheduled。 +8. 找到 MQ Producer。 +9. 找到 MQ Consumer。 +10. 找到 RestTemplate / Feign / HTTP Client。 +11. 找到 Netty / MQTT / TCP / UDP 入口。 +12. 找到 Logback 配置。 +13. 找到现有 MDC / ThreadLocal / RequestContext 实现。 +14. 检查项目是否已经存在类似 TraceId / RequestId 机制。 + +如果已经存在类似能力： + +> 优先重构和统一，不要重复造轮子。 + +--- + +# 二十二、实现原则 + +遵守： + +```text +低侵入 +高复用 +自动传播 +自动清理 +统一入口 +统一出口 +``` + +业务代码尽可能不需要修改。 + +理想效果： + +原来的代码： + +```java +log.info("开始处理设备"); +``` + +完全不用修改。 + +自动变成： + +```text +traceId=A 开始处理设备 +``` + +--- + +# 二十三、测试要求 + +实现完成后必须增加测试。 + +至少测试： + +### Test 1：HTTP + +```text +请求进入 +↓ +生成 TraceId +↓ +Controller +↓ +Service +↓ +日志 +↓ +Response Header +``` + +确认 TraceId 一致。 + +### Test 2：HTTP携带 TraceId + +```text +X-Trace-Id=A +``` + +确认后端仍然使用： + +```text +A +``` + +### Test 3：线程池 + +```text +Web traceId=A +↓ +ThreadPool +``` + +确认： + +```text +traceId=A +``` + +### Test 4：MQ + +```text +Producer traceId=A +↓ +MQ Header +↓ +Consumer +``` + +确认 Consumer： + +```text +traceId=A +``` + +### Test 5：MQ无 TraceId + +确认 Consumer 自动生成。 + +### Test 6：定时任务 + +确认： + +```text +TASK traceId=A +``` + +整个任务一致。 + +### Test 7：线程池污染 + +执行： + +```text +请求A +请求B +请求C +``` + +确认： + +```text +A ≠ B ≠ C +``` + +并且不会出现线程复用导致 TraceId 串线。 + +### Test 8：异常 + +异常发生后： + +```text +traceId +``` + +仍然存在于 ERROR 日志。 + +### Test 9：并发 + +同时执行大量请求： + +```text +100+ +1000+ +``` + +确认 TraceId 不串线。 + +--- + +# 二十四、最终输出要求 + +完成后请给我： + +## 1. 修改文件列表 + +例如： + +```text +新增： +xxx/TraceContext.java + +修改： +xxx/Logback.xml +xxx/xxxFilter.java +``` + +## 2. 架构说明 + +说明： + +```text +Web + ↓ +TraceContext + ↓ +MDC + +ThreadPool + ↓ +Context Propagation + +MQ + ↓ +Message Header + +HTTP + ↓ +X-Trace-Id +``` + +## 3. 使用示例 + +告诉我业务代码以后应该怎么写。 + +目标是： + +```java +log.info("开始处理"); +``` + +而不是让业务代码手动管理 TraceId。 + +## 4. 测试结果 + +说明所有测试是否通过。 + +## 5. 风险 + +明确指出： + +- 哪些场景已经支持 +- 哪些场景暂未支持 +- 是否存在 MDC 丢失风险 +- 是否存在线程池上下文污染风险 +- 是否存在 MQ TraceId 丢失风险 + +## 6. 后续演进建议 + +最后说明未来如何从： + +```text +TraceId + MDC +``` + +演进到： + +```text +TraceId ++ +SpanId ++ +ParentSpanId ++ +OpenTelemetry +``` + +但本次不要提前实现完整分布式链路追踪系统。 + +--- + +# 最终验收标准 + +只有同时满足以下条件才算完成： + +```text + traceId=A + │ + ┌───────┴───────┐ + │ │ + Web Task + │ + ▼ + Service + │ + ┌───────┴───────┐ + ▼ ▼ + ThreadPool MQ + │ │ + │ ▼ + │ Consumer + │ │ + └───────┬───────┘ + ▼ + AI/HTTP + │ + ▼ + DB +``` + +同一业务链可以通过： + +```text +traceId=A +``` + +在日志文件、IDEA Console、Kibana 中完整检索。 + +尤其必须保证： + +> **跨线程、跨线程池、跨 MQ、跨 HTTP 的 TraceId 不丢失；任务执行完成后上下文必须清理，不允许线程复用造成 TraceId 污染。** + +请先分析当前代码，再实施修改。 + +不要在没有检查项目现有实现的情况下直接创建重复的 TraceId / MDC / ThreadLocal 体系。 \ No newline at end of file diff --git a/fastbee-admin/src/main/java/com/fastbee/web/task/LogCleanTask.java b/fastbee-admin/src/main/java/com/fastbee/web/task/LogCleanTask.java new file mode 100644 index 00000000..a742b693 --- /dev/null +++ b/fastbee-admin/src/main/java/com/fastbee/web/task/LogCleanTask.java @@ -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; + } + } +} \ No newline at end of file diff --git a/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java b/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java index f2d1afb2..67f2d431 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java +++ b/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java @@ -1,6 +1,7 @@ package com.fastbee.common.config; import com.fastbee.common.constant.FastBeeConstant; +import com.fastbee.common.trace.MdcTaskDecorator; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -60,7 +61,11 @@ public class DeviceTask { return builder(FastBeeConstant.TASK.FUNCTION_INVOKE_TASK); } - /*内部消费线程*/ + /** + * 内部消费线程 + * 监听循环从消息恢复 TraceId 并逐条清理； + * TaskDecorator 用于让 @Async 消费方法继承监听线程的 TraceId + */ @Bean(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK) public Executor messageConsumeTaskExecutor() { return builder(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK); @@ -104,7 +109,7 @@ public class DeviceTask { } - /*组装线程池*/ + /*组装线程池（带 TaskDecorator，用于跨线程传播 TraceId）*/ private ThreadPoolTaskExecutor builder(String threadNamePrefix) { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(coreSize); @@ -115,6 +120,10 @@ public class DeviceTask { executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); //线程池名的前缀 executor.setThreadNamePrefix(threadNamePrefix); + + // 配置 TaskDecorator，自动传播 TraceId + executor.setTaskDecorator(new MdcTaskDecorator()); + executor.initialize(); return executor; } diff --git a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java index 89be5860..d8eebe75 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java +++ b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java @@ -15,6 +15,8 @@ public interface FastBeeConstant { String MQTT = "mqtt"; String PORT = "port"; String ADAPTER = "adapter"; + /** 链路追踪处理器名 */ + String TRACE = "trace"; String FRAMEDECODER = "frameDecoder"; String DISPATCHER = "dispatcher"; String DECODER = "decoder"; diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReplyBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReplyBo.java index e45f1ea8..c849844a 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReplyBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReplyBo.java @@ -8,6 +8,9 @@ import lombok.Data; @Data public class DeviceReplyBo { + /** 链路追踪ID */ + private String traceId; + /*设备下发消息id*/ private String messageId; /*标识符*/ diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReportBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReportBo.java index d4506ccd..edbce754 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReportBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceReportBo.java @@ -24,6 +24,9 @@ import java.util.List; @AllArgsConstructor public class DeviceReportBo { + /** 链路追踪ID */ + private String traceId; + /*设备编号或IMEI号*/ private String serialNumber; /*产品ID*/ diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceStatusBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceStatusBo.java index 6bcd9adc..68567cf2 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceStatusBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceStatusBo.java @@ -18,6 +18,10 @@ import java.util.Date; @Data @Builder public class DeviceStatusBo { + + /** 链路追踪ID */ + private String traceId; + /** * 设备客户端id */ diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceTestReportBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceTestReportBo.java index b7e3e7ff..cd43f370 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceTestReportBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/DeviceTestReportBo.java @@ -11,6 +11,9 @@ import java.util.List; @Data public class DeviceTestReportBo { + /** 链路追踪ID */ + private String traceId; + private Long productId; /** * 设备编号 diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/InvokeReqDto.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/InvokeReqDto.java index acbc8ed5..386c7457 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/InvokeReqDto.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/InvokeReqDto.java @@ -17,6 +17,9 @@ import java.util.Map; @Data public class InvokeReqDto { + /** 链路追踪ID */ + private String traceId; + @ApiModelProperty(value = "设备编号") private String serialNumber; diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/MQSendMessageBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/MQSendMessageBo.java index f302c489..c5d2f9cc 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/MQSendMessageBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/MQSendMessageBo.java @@ -14,6 +14,9 @@ import lombok.NoArgsConstructor; @NoArgsConstructor public class MQSendMessageBo { + /** 链路追踪ID */ + private String traceId; + public Boolean isShadow; public Long userId; public Long delay; diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeBo.java index fc6db91a..9ddf34ed 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeBo.java @@ -17,6 +17,9 @@ import java.math.BigDecimal; @JsonInclude(JsonInclude.Include.NON_EMPTY) public class OtaUpgradeBo { + /** 链路追踪ID */ + private String traceId; + // 设备编码 private String serialNumber; // 升级任务ID diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeDelayTask.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeDelayTask.java index ae5e3d50..96585439 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeDelayTask.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/ota/OtaUpgradeDelayTask.java @@ -17,6 +17,11 @@ import java.util.concurrent.TimeUnit; @Data public class OtaUpgradeDelayTask implements Delayed { + /** + * 链路追踪ID + */ + private String traceId; + /** * 固件id */ diff --git a/fastbee-common/src/main/java/com/fastbee/common/trace/MdcTaskDecorator.java b/fastbee-common/src/main/java/com/fastbee/common/trace/MdcTaskDecorator.java new file mode 100644 index 00000000..26758f30 --- /dev/null +++ b/fastbee-common/src/main/java/com/fastbee/common/trace/MdcTaskDecorator.java @@ -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 丢失问题 + * + *

使用方法： + * 

+ * ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+ * executor.setTaskDecorator(new MdcTaskDecorator());
+ *
+ * + *

工作原理： + *

    + *
  1. 在任务提交时，捕获当前线程的 TraceContext 快照
    2. + *
  2. 在任务执行时，恢复快照到新线程（MDC 由 TraceContext 统一同步）
    3. + *
  3. 在任务完成后，清理上下文防止线程池污染
    4. + *
+ * + *

说明：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 Callable decorate(Callable task) { + TraceSnapshot snapshot = TraceContext.snapshot(); + + return () -> { + if (snapshot != null) { + TraceContext.restore(snapshot); + } else { + TraceContext.getOrCreateTraceId(); + } + try { + return task.call(); + } finally { + TraceContext.clear(); + } + }; + } +} \ No newline at end of file diff --git a/fastbee-common/src/main/java/com/fastbee/common/trace/TraceContext.java b/fastbee-common/src/main/java/com/fastbee/common/trace/TraceContext.java new file mode 100644 index 00000000..7c807628 --- /dev/null +++ b/fastbee-common/src/main/java/com/fastbee/common/trace/TraceContext.java @@ -0,0 +1,270 @@ +package com.fastbee.common.trace; + +import lombok.Data; +import org.slf4j.MDC; + +import java.io.Serializable; + +/** + * 统一追踪上下文 + * 管理 traceId, tenantId, userId 等追踪信息 + * + *

核心特性： + *

+ * + *

使用示例： + * 

+ * // Web请求入口
+ * TraceContext.setTraceId(TraceIdGenerator.generate());
+ *
+ * // 异步任务（通过 TaskDecorator 自动传播）
+ * @Async
+ * public void asyncTask() {
+ *     log.info("处理中"); // 日志自动包含 traceId
+ * }
+ *
+ * // 手动传播（用于非 Spring 管理的线程池）
+ * TraceSnapshot snapshot = TraceContext.snapshot();
+ * executor.submit(() -> {
+ *     TraceContext.restore(snapshot);
+ *     try {
+ *         // 业务逻辑
+ *     } finally {
+ *         TraceContext.clear();
+ *     }
+ * });
+ *
+ * + * @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 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); + } +} \ No newline at end of file diff --git a/fastbee-common/src/main/java/com/fastbee/common/trace/TraceIdGenerator.java b/fastbee-common/src/main/java/com/fastbee/common/trace/TraceIdGenerator.java new file mode 100644 index 00000000..755780bb --- /dev/null +++ b/fastbee-common/src/main/java/com/fastbee/common/trace/TraceIdGenerator.java @@ -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 + * + *

格式：基于时间戳 + 序列号 + 随机数 + *

示例：752431931275776594 + * + *

特点： + *

+ * + * @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 保证线程安全 + * + *

位域分配（低到高）： + *

+ * + * @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; + } +} \ No newline at end of file diff --git a/fastbee-common/src/main/java/com/fastbee/common/utils/StringUtils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/StringUtils.java index 3b5b0e7b..ce9e00d4 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/utils/StringUtils.java +++ b/fastbee-common/src/main/java/com/fastbee/common/utils/StringUtils.java @@ -295,21 +295,19 @@ public class StringUtils extends org.apache.commons.lang3.StringUtils { /** * 判断给定的set列表中是否包含数组array 判断给定的数组array中是否包含给定的元素value * - * @param set 给定的集合 + * @param collection 给定的集合 * @param array 给定的数组 * @return boolean 结果 */ public static boolean containsAny(Collection collection, String... array) { - if (isEmpty(collection) || isEmpty(array)) { - return false; - } else { + if (!isEmpty(collection) && !isEmpty(array)) { for (String str : array) { if (collection.contains(str)) { return true; } } - return false; } + return false; } /** diff --git a/fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java index fdd7138c..c05442f7 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java +++ b/fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java @@ -1,188 +1,64 @@ package com.fastbee.common.utils; -import com.fastbee.common.utils.uuid.IdUtils; -import org.slf4j.MDC; +import com.fastbee.common.trace.TraceContext; +import com.fastbee.common.trace.TraceContext.TraceSnapshot; -import java.util.Map; import java.util.concurrent.Callable; -import java.util.concurrent.ThreadLocalRandom; /** * 日志追踪ID工具类 - * 统一生成TraceId的逻辑，使用IdWorker和ThreadLocalRandom + * 提供向后兼容的 API，内部委托给 TraceContext * * @author Sumuxi + * 建议直接使用 {@link TraceContext} */ public class TraceIdUtils { /** - * TraceId在MDC中的key - */ - public static final String TRACE_ID_KEY = "traceId"; - - /** - * IdWorker实例（单例模式） - */ - private static volatile IdUtils idWorker; - - /** - * 获取IdWorker实例（懒加载单例） + * 包装 Runnable：透传 TraceContext（MDC 由 TraceContext 统一同步） + * 用于非 Spring 管理的线程池（如 ScheduledExecutorService） * - * @return IdWorker实例 - */ - 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 则补齐；执行完自动清理/还原。 + * @param task Runnable 任务 + * @return 包装后的 Runnable */ public static Runnable wrapWithMdc(Runnable task) { - final Map captured = MDC.getCopyOfContextMap(); + final TraceSnapshot snapshot = TraceContext.snapshot(); + return () -> { - Map previous = MDC.getCopyOfContextMap(); + if (snapshot != null) { + TraceContext.restore(snapshot); + } else { + TraceContext.getOrCreateTraceId(); + } try { - if (captured == null || captured.isEmpty()) { - MDC.clear(); - } else { - MDC.setContextMap(captured); - } - ensureTraceId(); task.run(); } finally { - if (previous == null || previous.isEmpty()) { - MDC.clear(); - } else { - MDC.setContextMap(previous); - } + TraceContext.clear(); } }; } /** - * 包装 Callable：透传提交线程的 MDC，上下文中无 traceId 则补齐；执行完自动清理/还原。 + * 包装 Callable：透传 TraceContext（MDC 由 TraceContext 统一同步） + * 用于非 Spring 管理的线程池（如 ScheduledExecutorService） + * + * @param task Callable 任务 + * @return 包装后的 Callable */ public static Callable wrapWithMdc(Callable task) { - final Map captured = MDC.getCopyOfContextMap(); + final TraceSnapshot snapshot = TraceContext.snapshot(); + return () -> { - Map previous = MDC.getCopyOfContextMap(); + if (snapshot != null) { + TraceContext.restore(snapshot); + } else { + TraceContext.getOrCreateTraceId(); + } try { - if (captured == null || captured.isEmpty()) { - MDC.clear(); - } else { - MDC.setContextMap(captured); - } - ensureTraceId(); return task.call(); } finally { - if (previous == null || previous.isEmpty()) { - MDC.clear(); - } else { - MDC.setContextMap(previous); - } + TraceContext.clear(); } }; } - - /** - * 清除MDC中的TraceId - */ - public static void clearTraceId() { - MDC.remove(TRACE_ID_KEY); - } -} - +} \ No newline at end of file diff --git a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC8Utils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC8Utils.java index 48a26cc5..35e98b4c 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC8Utils.java +++ b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC8Utils.java @@ -25,12 +25,12 @@ public class CRC8Utils { */ public static byte calcCrc8_E5(byte[] data) { byte crc = 0; - for (int j = 0; j < data.length; j++) { - crc ^= data[j]; + for (byte datum : data) { + crc ^= datum; for (int i = 0; i < 8; i++) { if ((crc & 0x80) != 0) { crc = (byte) ((crc) << 1); - crc ^= 0xE5; + crc ^= (byte) 0xE5; } else { crc = (byte) ((crc) << 1); } diff --git a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java index c6ce73e0..1ac13ad5 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java +++ b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java @@ -78,8 +78,7 @@ public class ByteUtils { byte[] write = new byte[length]; int index = 0; - for (int i = 0; i < bytes.size(); i++) { - byte[] values = bytes.get(i); + for (byte[] values : bytes) { addBytes(write, values, index); index += values.length; } @@ -312,8 +311,9 @@ public class ByteUtils { if (StringUtils.isBlank(hexStr)) { return null; } - if (hexStr.length() % 2 != 0) {//长度为单数 - hexStr = "0" + hexStr;//前面补0 + // 长度为单数 + if (hexStr.length() % 2 != 0) { + hexStr = "0" + hexStr; // 前面补0 } char[] chars = hexStr.toCharArray(); diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/aspectj/ScheduledTraceAspect.java b/fastbee-framework/src/main/java/com/fastbee/framework/aspectj/ScheduledTraceAspect.java new file mode 100644 index 00000000..e4691c91 --- /dev/null +++ b/fastbee-framework/src/main/java/com/fastbee/framework/aspectj/ScheduledTraceAspect.java @@ -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(); + } + } +} \ No newline at end of file diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/config/OtaThreadPoolConfig.java b/fastbee-framework/src/main/java/com/fastbee/framework/config/OtaThreadPoolConfig.java index d460de10..6a4859e5 100644 --- a/fastbee-framework/src/main/java/com/fastbee/framework/config/OtaThreadPoolConfig.java +++ b/fastbee-framework/src/main/java/com/fastbee/framework/config/OtaThreadPoolConfig.java @@ -1,6 +1,7 @@ package com.fastbee.framework.config; import com.fastbee.common.constant.FastBeeConstant; +import com.fastbee.common.trace.MdcTaskDecorator; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -30,6 +31,10 @@ public class OtaThreadPoolConfig { executor.setThreadNamePrefix(threadNamePrefix); // 线程池对拒绝任务(无线程可用)的处理策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + + // 新增：配置 TaskDecorator，自动传播 TraceId + executor.setTaskDecorator(new MdcTaskDecorator()); + // 进行加载 executor.initialize(); return executor; diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/config/ThreadPoolConfig.java b/fastbee-framework/src/main/java/com/fastbee/framework/config/ThreadPoolConfig.java index cf18f388..6cbeeb7b 100644 --- a/fastbee-framework/src/main/java/com/fastbee/framework/config/ThreadPoolConfig.java +++ b/fastbee-framework/src/main/java/com/fastbee/framework/config/ThreadPoolConfig.java @@ -1,5 +1,6 @@ package com.fastbee.framework.config; +import com.fastbee.common.trace.MdcTaskDecorator; import com.fastbee.common.utils.Threads; import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.springframework.context.annotation.Bean; @@ -38,6 +39,10 @@ public class ThreadPoolConfig { executor.setKeepAliveSeconds(keepAliveSeconds); // 线程池对拒绝任务(无线程可用)的处理策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + + // 新增：配置 TaskDecorator，自动传播 TraceId + executor.setTaskDecorator(new MdcTaskDecorator()); + return executor; } diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/interceptor/TraceIdInterceptor.java b/fastbee-framework/src/main/java/com/fastbee/framework/interceptor/TraceIdInterceptor.java index b892f4ea..af9675c8 100644 --- a/fastbee-framework/src/main/java/com/fastbee/framework/interceptor/TraceIdInterceptor.java +++ b/fastbee-framework/src/main/java/com/fastbee/framework/interceptor/TraceIdInterceptor.java @@ -1,9 +1,9 @@ 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.web.servlet.HandlerInterceptor; - import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; @@ -16,24 +16,26 @@ import javax.servlet.http.HttpServletResponse; @Component public class TraceIdInterceptor implements HandlerInterceptor { + private static final String TRACE_ID_HEADER = "X-Trace-Id"; + @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { // 获取请求URI String requestURI = request.getRequestURI(); // 尝试从请求头中获取TraceId（支持分布式追踪） - String traceId = request.getHeader("X-Trace-Id"); + String traceId = request.getHeader(TRACE_ID_HEADER); // 如果请求头中没有TraceId，则生成一个新的 if (traceId == null || traceId.isEmpty()) { - traceId = TraceIdUtils.generateTraceIdString(); - } else { - // 如果请求头中有TraceId，设置到MDC - TraceIdUtils.setTraceId(traceId); + traceId = TraceIdGenerator.generate(); } + // 设置到 TraceContext（自动同步到 MDC） + TraceContext.setTraceId(traceId); + // 将TraceId放入响应头，方便前端获取 - response.setHeader("X-Trace-Id", traceId); + response.setHeader(TRACE_ID_HEADER, traceId); // 打印请求路径日志 org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(TraceIdInterceptor.class); @@ -44,8 +46,8 @@ public class TraceIdInterceptor implements HandlerInterceptor { @Override public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception { - // 请求完成后清除MDC中的TraceId，避免内存泄漏 - TraceIdUtils.clearTraceId(); + // 请求完成后清除 TraceContext，避免内存泄漏 + TraceContext.clear(); } } diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/manager/AsyncManager.java b/fastbee-framework/src/main/java/com/fastbee/framework/manager/AsyncManager.java index 7fe96dcd..5d18fff9 100644 --- a/fastbee-framework/src/main/java/com/fastbee/framework/manager/AsyncManager.java +++ b/fastbee-framework/src/main/java/com/fastbee/framework/manager/AsyncManager.java @@ -1,6 +1,7 @@ package com.fastbee.framework.manager; import com.fastbee.common.utils.Threads; +import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.common.utils.spring.SpringUtils; import java.util.TimerTask; @@ -36,10 +37,17 @@ public class AsyncManager { /** * 执行任务 * + *

包装任务以透传 TraceId：{@code scheduledExecutorService} 是裸 + * {@link java.util.concurrent.ScheduledThreadPoolExecutor}，不支持 + * {@code setTaskDecorator}，因此必须在提交点包装任务本身。 + * {@link TraceIdUtils#wrapWithMdc} 会捕获调用方线程的 TraceContext + * 快照并在 schedule-pool 线程内恢复，使登录/登出/操作日志等 + * 异步落库链路的日志携带同一 TraceId。 + * * @param task 任务 */ public void execute(TimerTask task) { - executor.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS); + executor.schedule(TraceIdUtils.wrapWithMdc(task), OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS); } /** diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java index e95e3dc0..3870fb90 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java @@ -99,7 +99,7 @@ public class DevicePropFetchConsumer { // 不进行格式校验，直接转换数据 byte[] commandData = ByteUtils.convertToHexByteArray(command); switch (serverType) { - //通过mqtt内部客户端 下发指令 + // 通过mqtt内部客户端 下发指令 case MQTT: publish(topic, commandData); log.info("=>MQTT-线程=[{}],设备=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), serialNumber, command, topic); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceAbstractQuartzJob.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceAbstractQuartzJob.java index c24fd805..b3b31169 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceAbstractQuartzJob.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceAbstractQuartzJob.java @@ -2,9 +2,10 @@ package com.fastbee.iot.data.job; import com.fastbee.common.constant.Constants; 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.StringUtils; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.common.utils.bean.BeanUtils; import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.iot.domain.DeviceJob; @@ -29,7 +30,7 @@ public abstract class DeviceAbstractQuartzJob implements Job { /** * 线程本地变量 */ - private static ThreadLocal threadLocal = new ThreadLocal<>(); + private static final ThreadLocal threadLocal = new ThreadLocal<>(); @Override 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)); try { before(context, deviceJob); - if (deviceJob != null) { - doExecute(context, deviceJob); - } + doExecute(context, deviceJob); after(context, deviceJob, null); } catch (Exception e) { log.error("任务执行异常 - ：", e); @@ -56,8 +55,8 @@ public abstract class DeviceAbstractQuartzJob implements Job { protected void before(JobExecutionContext context, DeviceJob deviceJob) { threadLocal.set(new Date()); - // 为设备定时任务生成TraceId，使用统一的生成逻辑 - TraceIdUtils.generateTraceId(); + // 为设备定时任务生成 TraceId + TraceContext.setTraceId(TraceIdGenerator.generate()); log.info("设备定时任务开始执行: {} - {}", deviceJob.getJobName(), deviceJob.getJobGroup()); } @@ -91,10 +90,6 @@ public abstract class DeviceAbstractQuartzJob implements Job { startTime = new Date(); } sysJobLog.setStartTime(startTime); - Date stopTime = new Date(); - sysJobLog.setStopTime(stopTime); - long runMs = stopTime.getTime() - startTime.getTime(); - sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒"); if (e != null) { sysJobLog.setStatus(Constants.FAIL); String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000); @@ -103,13 +98,18 @@ public abstract class DeviceAbstractQuartzJob implements Job { 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); - // 清除TraceId，避免内存泄漏 - TraceIdUtils.clearTraceId(); - log.info("设备定时任务执行完成: {} - {}, 耗时: {}ms", deviceJob.getJobName(), deviceJob.getJobGroup(), runMs); + + // 清除 TraceContext + TraceContext.clear(); } /** diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java index 62c5bb74..e45a4196 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java @@ -11,6 +11,7 @@ import org.quartz.JobExecutionContext; */ @DisallowConcurrentExecution public class DeviceDisallowConcurrentExecution extends DeviceAbstractQuartzJob { + @Override protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception { DeviceJobInvoke.invokeMethod(deviceJob); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java index 87788c0b..5f9af7c2 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java @@ -9,6 +9,7 @@ import org.quartz.JobExecutionContext; * @author sumuxi, 2026-04-16 */ public class DeviceJobExecution extends DeviceAbstractQuartzJob { + @Override protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception { DeviceJobInvoke.invokeMethod(deviceJob); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobInvoke.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobInvoke.java index 7f0f4332..cff7e810 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobInvoke.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobInvoke.java @@ -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 { if (deviceJob.getJobType() == 1) { - log.info("------------------------1、执行设备定时任务-----------------------------"); + log.info("1、[执行设备定时任务]"); List actions = JSON.parseArray(deviceJob.getActions(), Action.class); - List propertys = new ArrayList<>(); + List properties = new ArrayList<>(); List functions = new ArrayList<>(); - for (int i = 0; i < actions.size(); i++) { + for (Action action : actions) { ThingsModelSimpleItem model = new ThingsModelSimpleItem(); - model.setId(actions.get(i).getId()); - model.setValue(actions.get(i).getValue()); + model.setId(action.getId()); + model.setValue(action.getValue()); model.setRemark("设备定时"); - if (actions.get(i).getType() == 1) { - propertys.add(model); - } else if (actions.get(i).getType() == 2) { + if (action.getType() == 1) { + properties.add(model); + } else if (action.getType() == 2) { functions.add(model); } } // 发布属性 - if (propertys.size() > 0) { - messagePublish.publishProperty(deviceJob.getProductId(), deviceJob.getSerialNumber(), propertys, 0); + if (!properties.isEmpty()) { + messagePublish.publishProperty(deviceJob.getProductId(), deviceJob.getSerialNumber(), properties, 0); } // 发布功能 - if (functions.size() > 0) { + if (!functions.isEmpty()) { messagePublish.publishFunction(deviceJob.getProductId(), deviceJob.getSerialNumber(), functions, 0); } } else if (deviceJob.getJobType() == 3) { - log.info("------------------3、[定时执行场景联动]---------------------"); + log.info("3、[定时执行场景联动]"); Scene scene = sceneMapper.selectSceneBySceneId(deviceJob.getSceneId()); // 执行场景规则,异步非阻塞 SceneContext context = new SceneContext("", 0L, 0, null); String requestId = "scene/" + scene.getChainName(); flowExecutor.execute2FutureWithRid(scene.getChainName(), null, requestId, context); } else if (4 == deviceJob.getJobType()) { - log.info("------------------4、[定时执行场景运算型变量]---------------------"); + log.info("4、[定时执行场景运算型变量]"); String s = dataHandler.calculateSceneModelTagValue(deviceJob.getDatasourceId()); if (StringUtils.isEmpty(s)) { - log.info("------------------[定时执行场景运算型变量失败：+{}]---------------------", s); + log.info("[定时执行场景运算型变量失败：+{}]", s); } } else if (5 == deviceJob.getJobType()) { - log.info("----------------5、执行modbus轮训指令----------------------"); + log.info("5、[执行Modbus轮训指令]"); ModbusPollJob job = new ModbusPollJob(); job.setType(2); job.setDeviceJob(deviceJob); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/SyncDeviceStatusJob.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/SyncDeviceStatusJob.java index 6f113113..5b299e10 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/SyncDeviceStatusJob.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/SyncDeviceStatusJob.java @@ -38,7 +38,7 @@ public class SyncDeviceStatusJob { */ public void syncDeviceStatus() { if (enabled) { - //获取所有已激活并不是禁用的设备 + // 获取所有已激活并不是禁用的设备 List deviceStatusVOList = deviceService.selectDeviceActive(); if (!CollectionUtils.isEmpty(deviceStatusVOList)) { for (DeviceStatusVO statusVO : deviceStatusVOList) { diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceOtherListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceOtherListen.java index 0b84e340..8679a1af 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceOtherListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceOtherListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceReportBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.DeviceOtherMsgConsumer; import com.fastbee.mq.queue.DeviceOtherQueue; import lombok.extern.slf4j.Slf4j; @@ -26,7 +27,15 @@ public class DeviceOtherListen { while (true) { try { DeviceReportBo reportBo = DeviceOtherQueue.take(); - otherMsgConsumer.consume(reportBo); + + // 恢复 TraceId + TraceContext.ensureTraceId(reportBo.getTraceId()); + + try { + otherMsgConsumer.consume(reportBo); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("=>emq数据转发异常"); } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DevicePropFetchListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DevicePropFetchListen.java index 167a9b41..06388c3c 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DevicePropFetchListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DevicePropFetchListen.java @@ -1,6 +1,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.DevicePropFetchConsumer; import com.fastbee.iot.model.modbus.ModbusPollJob; import com.fastbee.mq.queue.DevicePropFetchQueue; @@ -27,9 +28,20 @@ public class DevicePropFetchListen { while (true) { try { ModbusPollJob take = DevicePropFetchQueue.take(); - devicePropFetchConsumer.consume(take); + + // 恢复 TraceId + TraceContext.ensureTraceId(take.getTraceId()); + + try { + devicePropFetchConsumer.consume(take); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("=>设备属性获取异常", e); + } finally { + // 确保任何情况下都清理 + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReplyListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReplyListen.java index fee5854f..9109c799 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReplyListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReplyListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceReportBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.DeviceReplyMsgConsumer; import com.fastbee.mq.queue.DeviceReplyQueue; import lombok.extern.slf4j.Slf4j; @@ -27,10 +28,21 @@ public class DeviceReplyListen { try { /*读队列消息*/ DeviceReportBo reportBo = DeviceReplyQueue.take(); - /*处理消息*/ - deviceReplyMsgHandler.consume(reportBo); + + // 恢复 TraceId + TraceContext.ensureTraceId(reportBo.getTraceId()); + + try { + /*处理消息*/ + deviceReplyMsgHandler.consume(reportBo); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("=>设备回调消息监听异常", e); + } finally { + // 确保任何情况下都清理上下文 + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReportListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReportListen.java index caad7495..fa28bb03 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReportListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReportListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceReportBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.DeviceReportMsgConsumer; import com.fastbee.mq.queue.DeviceReportQueue; import lombok.extern.slf4j.Slf4j; @@ -27,10 +28,21 @@ public class DeviceReportListen { try { /*取出数据*/ DeviceReportBo reportBo = DeviceReportQueue.take(); - /*处理数据*/ - reportMsgConsumer.consume(reportBo); + + // 恢复 TraceId + TraceContext.ensureTraceId(reportBo.getTraceId()); + + try { + /*处理数据*/ + reportMsgConsumer.consume(reportBo); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("=>设备上报数据监听异常", e); + } finally { + // 确保任何情况下都清理上下文 + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceStatusListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceStatusListen.java index 6252c0cb..07bdf6b9 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceStatusListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceStatusListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceStatusBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.DeviceStatusConsumer; import com.fastbee.mq.queue.DeviceStatusQueue; import lombok.extern.slf4j.Slf4j; @@ -26,10 +27,20 @@ public class DeviceStatusListen { try { while (true) { DeviceStatusBo status = DeviceStatusQueue.take(); - deviceStatusConsumer.consume(status); + + // 恢复 TraceId + TraceContext.ensureTraceId(status.getTraceId()); + + try { + deviceStatusConsumer.consume(status); + } finally { + TraceContext.clear(); + } } } catch (Exception e) { log.error("设备状态监听错误", e); + } finally { + TraceContext.clear(); } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceTestListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceTestListen.java index 1b4de8a7..c3cd81e8 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceTestListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceTestListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceTestReportBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.DeviceTestConsumer; import com.fastbee.mq.queue.DeviceTestQueue; import lombok.extern.slf4j.Slf4j; @@ -26,9 +27,19 @@ public class DeviceTestListen { try { DeviceTestReportBo take = DeviceTestQueue.take(); - deviceTestConsumer.consume(take); + // 恢复 TraceId + TraceContext.ensureTraceId(take.getTraceId()); + + try { + deviceTestConsumer.consume(take); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("=>emq数据转发异常"); + } finally { + // 确保任何情况下都清理上下文 + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/FunctionInvokeListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/FunctionInvokeListen.java index a631bd22..aa4473ae 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/FunctionInvokeListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/FunctionInvokeListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.MQSendMessageBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.consumer.FunctionInvokeConsumer; import com.fastbee.mq.queue.FunctionInvokeQueue; import lombok.extern.slf4j.Slf4j; @@ -26,9 +27,20 @@ public class FunctionInvokeListen { while (true) { try { MQSendMessageBo sendBo = FunctionInvokeQueue.take(); - functionInvokeConsumer.handler(sendBo); + + // 恢复 TraceId + TraceContext.ensureTraceId(sendBo.getTraceId()); + + try { + functionInvokeConsumer.handler(sendBo); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("=>下发服务消费异常", e); + } finally { + // 确保任何情况下都清理上下文 + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpGradeListener.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpGradeListener.java index 56a6ded2..94a7d13c 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpGradeListener.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpGradeListener.java @@ -4,6 +4,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.domain.entity.SysMenu; import com.fastbee.common.core.mq.ota.OtaUpgradeDelayTask; +import com.fastbee.common.trace.TraceContext; import com.fastbee.common.utils.StringUtils; import com.fastbee.iot.data.service.IOtaUpgradeService; import com.fastbee.mq.queue.DelayUpgradeQueue; @@ -48,12 +49,22 @@ public class UpGradeListener implements ApplicationRunner { OtaUpgradeDelayTask task = DelayUpgradeQueue.task(); if (StringUtils.isNotNull(task)) { Date startTime = task.getStartTime(); - otaUpgradeService.upgrade(task); - log.info("=>开始OTA升级时间{}", startTime); + + // 恢复 TraceId + TraceContext.ensureTraceId(task.getTraceId()); + + try { + otaUpgradeService.upgrade(task); + log.info("=>开始OTA升级时间{}", startTime); + } finally { + TraceContext.clear(); + } } continue; } catch (Exception e) { log.error("=>OTA升级监听异常", e); + } finally { + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpgradeListen.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpgradeListen.java index e7ea9a22..eb59af71 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpgradeListen.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpgradeListen.java @@ -2,6 +2,7 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.ota.OtaUpgradeBo; +import com.fastbee.common.trace.TraceContext; import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.mq.queue.OtaUpgradeQueue; import lombok.extern.slf4j.Slf4j; @@ -28,10 +29,20 @@ public class UpgradeListen { try { /*获取队列中的OTA升级消息*/ OtaUpgradeBo upgradeBo = OtaUpgradeQueue.take(); - // OTA升级处理 - functionSendService.upGradeOTA(upgradeBo); + + // 恢复 TraceId + TraceContext.ensureTraceId(upgradeBo.getTraceId()); + + try { + // OTA升级处理 + functionSendService.upGradeOTA(upgradeBo); + } finally { + TraceContext.clear(); + } } catch (Exception e) { log.error("->OTA消息监听异常", e); + } finally { + TraceContext.clear(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java index 687a9dbb..fe52ccfb 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java @@ -47,6 +47,10 @@ public class DeviceJobServiceImpl implements IDeviceJobService { @PostConstruct public void init() throws SchedulerException, TaskException { scheduler.clear(); + // 打印实际类型 +// log.info("Scheduler实际类型: {}", scheduler.getClass().getName()); +// log.info("Scheduler元数据: {}", scheduler.getMetaData().getSummary()); +// log.info("Scheduler class: {}", scheduler.getClass()); // 设备定时任务 List jobList = jobMapper.selectJobAll(); for (DeviceJob deviceJob : jobList) { diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java index 009c6e8b..20f45d25 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java @@ -222,7 +222,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { /** * 变量读取 * - * @param readVO + * @param readVO 前端入参 */ @Override public void readVariableValue(VariableReadVO readVO) { @@ -233,7 +233,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { deviceAndProtocol.getProtocolCode().equals(FastBeeConstant.PROTOCOL.ModbusTcp))) throw new ServiceException("非modbus协议请先配置主动采集协议"); Long productId = deviceAndProtocol.getProductId(); - //如果是子设备，则获取网关子设备的产品id和设备编号 + // 如果是子设备，则获取网关子设备的产品id和设备编号 Integer deviceType = deviceAndProtocol.getDeviceType(); if (deviceType == DeviceType.SUB_GATEWAY.getCode()) { serialNumber = deviceAndProtocol.getGwSerialNumber(); @@ -242,7 +242,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { Integer type = readVO.getType(); type = Objects.isNull(type) ? 1 : type; if (type == 1) { - //单个变量获取 + // 单个变量获取 String identifier = readVO.getIdentifier(); ThingsModelValueItem thingModels = itslCache.getSingleThingModels(deviceAndProtocol.getProductId(), identifier); ModbusConfig config = thingModels.getConfig(); @@ -275,7 +275,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { job.setType(1); MessageProducer.sendPropFetch(job); } else { - //读取当前设备的所有变量,这里读取所有的，判断传递的设备是网关设备、网关子设备、直连设备 + // 读取当前设备的所有变量,这里读取所有的，判断传递的设备是网关设备、网关子设备、直连设备 DeviceType code = DeviceType.transfer(deviceType); List modbusJobList = modbusJobService.selectDevicesJobByDeviceType(code, readVO.getSerialNumber()); List commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList()); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/FirmwareTaskServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/FirmwareTaskServiceImpl.java index abe92afd..e522d24a 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/FirmwareTaskServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/FirmwareTaskServiceImpl.java @@ -6,6 +6,7 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.fastbee.common.core.mq.ota.OtaUpgradeDelayTask; import com.fastbee.common.enums.OTAUpgrade; import com.fastbee.common.exception.ServiceException; +import com.fastbee.common.trace.TraceContext; import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.MessageUtils; import com.fastbee.common.utils.StringUtils; @@ -140,7 +141,9 @@ public class FirmwareTaskServiceImpl extends ServiceImpl + + + DELETE FROM notify_log + WHERE create_time < DATE_SUB(NOW(), INTERVAL #{days} DAY) + diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformDeviceMappingController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformDeviceMappingController.java index ed857816..306bae39 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformDeviceMappingController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformDeviceMappingController.java @@ -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 parentIds) { + return success(subMappingService.selectByParentIds(parentIds)); + } + @PreAuthorize("@ss.hasAnyPermi('platform:deviceMapping:query,platform:deviceMapping:list')") @GetMapping("/{id}/sub-mappings") @ApiOperation("查询子设备映射列表") diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/system/LogCleanController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/system/LogCleanController.java new file mode 100644 index 00000000..5d66c39c --- /dev/null +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/system/LogCleanController.java @@ -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()); + } + } +} \ No newline at end of file diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/tool/ToolController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/tool/ToolController.java index 5254dd68..337dd926 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/tool/ToolController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/tool/ToolController.java @@ -254,20 +254,19 @@ public class ToolController extends BaseController { } } catch (Exception ex) { - ex.printStackTrace(); log.error("发生错误：{}", ex.getMessage(), ex); } } @ApiOperation("mqtt钩子处理") @PostMapping("/mqtt/webhookv5") - public void webHookProcessv5(@RequestBody JSONObject json) { + public void webHookProcessV5(@RequestBody JSONObject json) { try { String clientid = json.getString("clientid"); String event = json.getString("event"); String peername = json.getString("peername"); // 类似 127.0.0.1:8812 String ipAddress = peername; - if (peername.indexOf(":") != -1) { + if (peername.contains(":")) { ipAddress = peername.substring(0, peername.indexOf(":")); } log.info("webhook:{},clientid:{},peername:{}", event, clientid, peername); @@ -310,18 +309,17 @@ public class ToolController extends BaseController { // 影子模式，发布属性和功能 if (device.getIsShadow() == 1) { ThingsModelShadow shadow = deviceService.getDeviceShadowThingsModel(device); - if (shadow.getProperties().size() > 0) { + if (!shadow.getProperties().isEmpty()) { 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); } } } } catch (Exception ex) { - ex.printStackTrace(); - log.error("发生错误：" + ex.getMessage()); + log.error("发生错误：{}", ex.getMessage()); } } diff --git a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/controller/SysJobController.java b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/controller/SysJobController.java index b6f01656..bf0ddd6e 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/controller/SysJobController.java +++ b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/controller/SysJobController.java @@ -83,9 +83,9 @@ public class SysJobController extends BaseController { return error(StringUtils.format(MessageUtils.message("job.add.failed.cron.not.valid"), job.getJobName())); } else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) { 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())); - } 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())); } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) { 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())); } else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) { 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())); - } 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())); } else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) { return error(StringUtils.format(MessageUtils.message("job.update.failed.string.error"), job.getJobName())); diff --git a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/mapper/SysJobLogMapper.java b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/mapper/SysJobLogMapper.java index 8b35eefe..cf40842d 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/mapper/SysJobLogMapper.java +++ b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/mapper/SysJobLogMapper.java @@ -61,4 +61,12 @@ public interface SysJobLogMapper { * 清空任务日志 */ public void cleanJobLog(); + + /** + * 删除指定天数前的任务日志 + * + * @param days 保留天数 + * @return 删除记录数 + */ + public int deleteJobLogBeforeDays(int days); } diff --git a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/ISysJobLogService.java b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/ISysJobLogService.java index 50893074..a0bea432 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/ISysJobLogService.java +++ b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/ISysJobLogService.java @@ -53,4 +53,11 @@ public interface ISysJobLogService { * 清空任务日志 */ public void cleanJobLog(); + + /** + * 清理指定天数前的任务日志 + * + * @param days 保留天数 + */ + public void cleanJobLog(int days); } diff --git a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/impl/SysJobLogServiceImpl.java b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/impl/SysJobLogServiceImpl.java index a5dfbe5e..8ba0f9dc 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/impl/SysJobLogServiceImpl.java +++ b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/service/impl/SysJobLogServiceImpl.java @@ -78,4 +78,14 @@ public class SysJobLogServiceImpl implements ISysJobLogService { public void cleanJobLog() { jobLogMapper.cleanJobLog(); } + + /** + * 清理指定天数前的任务日志 + * + * @param days 保留天数 + */ + @Override + public void cleanJobLog(int days) { + jobLogMapper.deleteJobLogBeforeDays(days); + } } diff --git a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/AbstractQuartzJob.java b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/AbstractQuartzJob.java index 6789a91b..30c6f2fe 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/AbstractQuartzJob.java +++ b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/AbstractQuartzJob.java @@ -2,9 +2,10 @@ package com.fastbee.quartz.util; import com.fastbee.common.constant.Constants; 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.StringUtils; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.common.utils.bean.BeanUtils; import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.quartz.domain.SysJob; @@ -29,7 +30,7 @@ public abstract class AbstractQuartzJob implements Job { /** * 线程本地变量 */ - private static ThreadLocal threadLocal = new ThreadLocal<>(); + private static final ThreadLocal threadLocal = new ThreadLocal<>(); @Override 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)); try { before(context, sysJob); - if (sysJob != null) { - doExecute(context, sysJob); - } + doExecute(context, sysJob); after(context, sysJob, null); } catch (Exception e) { log.error("任务执行异常 - ：", e); @@ -56,8 +55,8 @@ public abstract class AbstractQuartzJob implements Job { protected void before(JobExecutionContext context, SysJob sysJob) { threadLocal.set(new Date()); - // 为定时任务生成TraceId，使用统一的生成逻辑 - TraceIdUtils.generateTraceId(); + // 为定时任务生成 TraceId + TraceContext.setTraceId(TraceIdGenerator.generate()); log.info("定时任务开始执行: {} - {}", sysJob.getJobName(), sysJob.getJobGroup()); } @@ -77,9 +76,6 @@ public abstract class AbstractQuartzJob implements Job { sysJobLog.setJobGroup(sysJob.getJobGroup()); sysJobLog.setInvokeTarget(sysJob.getInvokeTarget()); sysJobLog.setStartTime(startTime); - sysJobLog.setStopTime(new Date()); - long runMs = sysJobLog.getStopTime().getTime() - sysJobLog.getStartTime().getTime(); - sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒"); if (e != null) { sysJobLog.setStatus(Constants.FAIL); String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000); @@ -87,14 +83,17 @@ public abstract class AbstractQuartzJob implements Job { } else { 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); - // 清除TraceId，避免内存泄漏 - TraceIdUtils.clearTraceId(); - log.info("定时任务执行完成: {} - {}, 耗时: {}ms", sysJob.getJobName(), sysJob.getJobGroup(), runMs); + + // 清除 TraceContext + TraceContext.clear(); + } /** diff --git a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/JobInvokeUtil.java b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/JobInvokeUtil.java index 1e78616a..02a36297 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/JobInvokeUtil.java +++ b/fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/JobInvokeUtil.java @@ -99,8 +99,8 @@ public class JobInvokeUtil { } String[] methodParams = methodStr.split(",(?=([^\"']*[\"'][^\"']*[\"'])*[^\"']*$)"); List classs = new LinkedList<>(); - for (int i = 0; i < methodParams.length; i++) { - String str = StringUtils.trimToEmpty(methodParams[i]); + for (String methodParam : methodParams) { + String str = StringUtils.trimToEmpty(methodParam); // String字符串类型，以'或"开头 if (StringUtils.startsWithAny(str, "'", "\"")) { 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()]; int index = 0; for (Object[] os : methodParams) { - classs[index] = (Object) os[0]; + classs[index] = os[0]; index++; } return classs; diff --git a/fastbee-plugs/fastbee-quartz/src/main/resources/mapper/quartz/SysJobLogMapper.xml b/fastbee-plugs/fastbee-quartz/src/main/resources/mapper/quartz/SysJobLogMapper.xml index 1a3e279e..9aafb989 100644 --- a/fastbee-plugs/fastbee-quartz/src/main/resources/mapper/quartz/SysJobLogMapper.xml +++ b/fastbee-plugs/fastbee-quartz/src/main/resources/mapper/quartz/SysJobLogMapper.xml @@ -77,6 +77,11 @@ truncate table sys_job_log + + DELETE FROM sys_job_log + WHERE create_time < DATE_SUB(NOW(), INTERVAL #{days} DAY) + + insert into sys_job_log( job_log_id, diff --git a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/model/ModbusRtu.java b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/model/ModbusRtu.java index 32dcb007..c87d6283 100644 --- a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/model/ModbusRtu.java +++ b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/model/ModbusRtu.java @@ -47,7 +47,7 @@ public class ModbusRtu extends Message { protected int start; @Column(length = 1, version = {0x80, 0x81}, desc = "标识位,实例设备是0x80") protected int mId; - @Column(length = 6, version = 0x80, desc = "MAC地址,6个字节") + @Column(length = 6, version = {0x80, 0x81}, desc = "MAC地址,6个字节") protected String mac; @Column(length = 1, version = {0x80, 0x81}, desc = "结尾,如：7E") protected int end; diff --git a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/pak/ModbusEndPoint.java b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/pak/ModbusEndPoint.java index aad04a54..417c0efb 100644 --- a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/pak/ModbusEndPoint.java +++ b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/pak/ModbusEndPoint.java @@ -39,7 +39,14 @@ public class ModbusEndPoint { @PakMapping(types = 心跳包) 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); + } } } diff --git a/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java b/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java index 703e36c2..6551a042 100644 --- a/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java +++ b/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java @@ -1,5 +1,6 @@ package com.fastbee.record.config; +import com.fastbee.common.trace.MdcTaskDecorator; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @@ -52,6 +53,10 @@ public class ThreadPoolTaskConfig { // 线程池对拒绝任务的处理策略 // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + + // 新增：配置 TaskDecorator，自动传播 TraceId + executor.setTaskDecorator(new MdcTaskDecorator()); + // 初始化 executor.initialize(); return executor; diff --git a/fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/TraceHandler.java b/fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/TraceHandler.java new file mode 100644 index 00000000..557585d2 --- /dev/null +++ b/fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/TraceHandler.java @@ -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 链路追踪处理器 + * + *

放置在协议 pipeline 的最前端（编解码器之前），为每条入站报文生成独立的 TraceId， + * 使解码、分发、业务处理全链路的日志都携带同一个 TraceId。 + * + *

注意：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(); + } + } +} \ No newline at end of file diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java index 6a8ba7bb..d5ddb77a 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java @@ -1,5 +1,6 @@ package com.fastbee.server; +import com.fastbee.base.core.hanler.TraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.server.config.NettyConfig; import com.fastbee.server.handler.*; @@ -61,9 +62,11 @@ public class TCPServer extends Server { @Override protected void initChannel(NioSocketChannel channel) throws Exception { channel.pipeline() + // 第一步、链路追踪（必须位于编解码器之前，让解码日志也带 TraceId） + .addLast(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE) // 设置心跳时间 .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime)) - // 第一步、添加粘包处理器 + // 第二步、添加粘包处理器 .addLast(FastBeeConstant.SERVER.FRAMEDECODER, new ProtocolSniffingDecoder(config.maxFrameLength)) // 第二步、添加消息适配器 .addLast(FastBeeConstant.SERVER.ADAPTER, adapter) diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/UDPServer.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/UDPServer.java index 35ffb17f..ae05b9dd 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/UDPServer.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/UDPServer.java @@ -1,5 +1,6 @@ package com.fastbee.server; +import com.fastbee.base.core.hanler.TraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.server.config.NettyConfig; import com.fastbee.server.handler.DispatcherHandler; @@ -52,6 +53,8 @@ public class UDPServer extends Server { @Override protected void initChannel(NioDatagramChannel channel) throws Exception { channel.pipeline() + // 链路追踪（位于编解码器之前，让解码日志也带 TraceId） + .addLast(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE) .addLast(FastBeeConstant.SERVER.ADAPTER, adapter) .addLast(FastBeeConstant.SERVER.DECODER, decoder) .addLast(FastBeeConstant.SERVER.ENCODER, encoder) diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java index 8b8abca0..008e7564 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java @@ -13,6 +13,7 @@ import com.fastbee.common.core.mq.DeviceTestReportBo; import com.fastbee.common.core.protocol.Message; import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem; import com.fastbee.common.enums.ServerType; +import com.fastbee.common.trace.TraceContext; import com.fastbee.common.utils.DateUtils; import com.fastbee.mq.producer.MessageProducer; import io.netty.channel.ChannelHandler; @@ -61,6 +62,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { s.increment(); } + /*TraceId 由 pipeline 头部的 TraceHandler 统一生成与清理*/ Packet packet = (Packet) msg; Message request = packet.message; // 判断是否注册 @@ -127,7 +129,16 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { } } else { 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 { channelRead0(ctx, packet, handler); } diff --git a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java index b6d5d146..6708dc56 100644 --- a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java +++ b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java @@ -55,6 +55,8 @@ public class MqttMessageDelegate { log.error("=>客户端未连接"); throw new ServiceException("客户端未连接"); } + + /*TraceId 由 pipeline 头部的 TraceHandler 统一生成与清理*/ Optional.of(processor.get(messageType)) .ifPresent(mqttHandler -> mqttHandler.handler(ctx, message)); } diff --git a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java index f81f9342..7fe40f53 100644 --- a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java +++ b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java @@ -1,5 +1,6 @@ package com.fastbee.mqtt.server; +import com.fastbee.base.core.hanler.TraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter; import com.fastbee.server.Server; @@ -54,6 +55,8 @@ public class MqttServer extends Server { channel.pipeline() .addFirst(FastBeeConstant.SERVER.IDLE , 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.ENCODER, MqttEncoder.INSTANCE) .addLast(messageAdapter); diff --git a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java index 33ea603d..a7962116 100644 --- a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java +++ b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java @@ -1,5 +1,6 @@ package com.fastbee.mqtt.server; +import com.fastbee.base.core.hanler.TraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.mqtt.codec.WebSocketMqttCodec; import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter; @@ -49,6 +50,8 @@ public class WebSocketServer extends Server { ch.pipeline() .addFirst(FastBeeConstant.WS.HEART_BEAT , new IdleStateHandler(0, 0, 70)) + /*链路追踪（位于编解码器之前，让解码日志也带 TraceId）*/ + .addFirst(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE) /*http请求响应*/ .addLast(FastBeeConstant.WS.HTTP_SERVER_CODEC, new HttpServerCodec()) /*聚合header与body组成完整的Http请求，最大数据量为1Mb*/ diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java index ac8d78ab..a07e51a7 100644 --- a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java @@ -1,5 +1,6 @@ package com.fastbee.sip.conf; +import com.fastbee.common.trace.MdcTaskDecorator; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @@ -26,6 +27,8 @@ public class ThreadPoolTaskConfig { executor.setKeepAliveSeconds(keepAliveTime); executor.setThreadNamePrefix(threadNamePrefix); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + // 配置 TaskDecorator，自动传播 TraceId + executor.setTaskDecorator(new MdcTaskDecorator()); executor.initialize(); return executor; } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/AlertLogMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/AlertLogMapper.java index 93ad999b..7a840871 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/AlertLogMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/AlertLogMapper.java @@ -115,4 +115,12 @@ public interface AlertLogMapper { * @return com.fastbee.iot.model.AlertCountVO */ List countAlertLevel(DataCenterParam dataCenterParam); + + /** + * 删除指定天数前的告警日志 + * + * @param days 保留天数 + * @return 删除记录数 + */ + int deleteAlertLogBeforeDays(@Param("days") int days); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java index b89ad082..6c3b5ea3 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java @@ -25,4 +25,12 @@ public interface EventLogMapper extends BaseMapperX { Long selectEventLogCount(@Param("userId") Long userId); + /** + * 删除指定天数前的日志 + * + * @param days 保留天数 + * @return 删除记录数 + */ + int deleteEventLogBeforeDays(@Param("days") int days); + } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/modbus/ModbusPollJob.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/modbus/ModbusPollJob.java index 237d444e..0c34e4f5 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/modbus/ModbusPollJob.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/modbus/ModbusPollJob.java @@ -11,6 +11,9 @@ import lombok.Data; @Data public class ModbusPollJob { + /** 链路追踪ID */ + private String traceId; + private ModbusPollMsg pollMsg; private DeviceJob deviceJob; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformDeviceSubMappingMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformDeviceSubMappingMapper.java index 588134cb..0ef9ba03 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformDeviceSubMappingMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformDeviceSubMappingMapper.java @@ -10,6 +10,9 @@ public interface PlatformDeviceSubMappingMapper extends BaseMapperX selectByParentId(@Param("parentMappingId") Long parentMappingId); + /** 批量查询多个父映射下的所有子映射（IN 查询，列表页展开行使用） */ + List selectByParentIds(@Param("parentIds") List parentIds); + PlatformDeviceSubMapping selectByParentAndSerial(@Param("parentMappingId") Long parentMappingId, @Param("subSerialNumber") String subSerialNumber); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceSubMappingService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceSubMappingService.java index 9eeacffe..23c48331 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceSubMappingService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceSubMappingService.java @@ -4,12 +4,16 @@ import com.baomidou.mybatisplus.extension.service.IService; import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; import java.util.List; +import java.util.Map; public interface IPlatformDeviceSubMappingService extends IService { /** 查询父映射下的所有子映射 */ List selectByParentId(Long parentMappingId); + /** 批量查询多个父映射下的子映射，按 parentMappingId 分组返回（列表页展开行一次拉取，避免 N 次请求） */ + Map> selectByParentIds(List parentIds); + /** 根据子设备序列号查找子映射（推送时使用） */ List selectBySubSerialNumber(String subSerialNumber); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java index d4bb4c03..895a1e54 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java @@ -3,6 +3,8 @@ package com.fastbee.iot.platform.service.impl; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; 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.flow.FlowContext; import com.fastbee.iot.platform.flow.PlatformFlowExecutor; @@ -21,6 +23,7 @@ import org.springframework.util.StringUtils; import java.util.ArrayList; import java.util.Date; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.stream.Collectors; @@ -41,6 +44,8 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl selectList(PlatformDeviceMapping query) { @@ -58,7 +63,25 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl rows = this.list(wrapper); + // 批量补全设备名称：device_name 为冗余字段，旧数据可能为空，按 deviceId 一次 IN 查询回填，避免前端逐个查询 + List deviceIds = rows.stream() + .filter(m -> m.getDeviceId() != null && !StringUtils.hasText(m.getDeviceName())) + .map(PlatformDeviceMapping::getDeviceId) + .distinct() + .collect(Collectors.toList()); + if (!deviceIds.isEmpty()) { + List devices = deviceMapper.selectList(Wrappers.lambdaQuery() + .in(Device::getDeviceId, deviceIds)); + Map 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 diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceSubMappingServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceSubMappingServiceImpl.java index 8e331fbe..873b6e0d 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceSubMappingServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceSubMappingServiceImpl.java @@ -8,7 +8,11 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; 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.Map; import java.util.Set; import java.util.stream.Collectors; @@ -24,6 +28,25 @@ public class PlatformDeviceSubMappingServiceImpl extends ServiceImpl> selectByParentIds(List parentIds) { + if (parentIds == null || parentIds.isEmpty()) { + return Collections.emptyMap(); + } + List all = subMappingMapper.selectByParentIds(parentIds); + Map> 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 public List selectBySubSerialNumber(String subSerialNumber) { return subMappingMapper.selectBySubSerialNumber(subSerialNumber); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IAlertLogService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IAlertLogService.java index 2aa708ca..311277e6 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IAlertLogService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IAlertLogService.java @@ -107,4 +107,11 @@ public interface IAlertLogService { * @return com.fastbee.iot.model.AlertCountVO */ List countAlertLevel(DataCenterParam dataCenterParam); + + /** + * 清理告警日志 + * + * @param days 保留天数 + */ + void cleanAlertLog(int days); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IEventLogService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IEventLogService.java index 5807e9db..0f94f59d 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IEventLogService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IEventLogService.java @@ -37,4 +37,11 @@ public interface IEventLogService extends IService { */ List listHistory(EventLog eventLog); + /** + * 清理事件日志 + * + * @param days 保留天数 + */ + void cleanEventLog(int days); + } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/AlertLogServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/AlertLogServiceImpl.java index f309e7ca..b4d9f367 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/AlertLogServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/AlertLogServiceImpl.java @@ -199,4 +199,14 @@ public class AlertLogServiceImpl implements IAlertLogService { public List countAlertLevel(DataCenterParam dataCenterParam) { return alertLogMapper.countAlertLevel(dataCenterParam); } + + /** + * 清理告警日志 + * + * @param days 保留天数 + */ + @Override + public void cleanAlertLog(int days) { + alertLogMapper.deleteAlertLogBeforeDays(days); + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DataRepairServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DataRepairServiceImpl.java index f7a43b12..6902bd9c 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DataRepairServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DataRepairServiceImpl.java @@ -210,7 +210,7 @@ public class DataRepairServiceImpl implements IDataRepairService { public void executeDataRepair(String serialNumber, boolean isShadow, List currentLogs) { if (CollectionUtils.isEmpty(currentLogs) || isShadow) { - log.debug("无数据或为影子模式，不执行数据修复: serialNumber={}, isShadow={}", serialNumber, isShadow); + //log.debug("无数据或为影子模式，不执行数据修复: serialNumber={}, isShadow={}", serialNumber, isShadow); return; } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java index cec4da61..3ec87d94 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java @@ -648,8 +648,7 @@ public class DeviceServiceImpl implements IDeviceService { */ @Override @DSTransactional - public List reportDeviceThingsModelValue(ThingsModelValuesInput input, int type, - boolean isShadow) { + public List reportDeviceThingsModelValue(ThingsModelValuesInput input, int type, boolean isShadow) { String serialNumber = input.getDeviceNumber(); String key = RedisKeyBuilder.buildTSLVCacheKey(input.getProductId(), serialNumber); Map maps = new HashMap<>(); @@ -675,6 +674,9 @@ public class DeviceServiceImpl implements IDeviceService { ThingsModelSimpleItem::getValue, (oldValue, newValue) -> oldValue)); } + // 查询设备信息 + Device device = this.selectDeviceBySerialNumber(serialNumber); + for (ThingsModelSimpleItem item : thingsModelValueRemarkItem) { String identity = item.getId(); if (identity.startsWith("array_")) { @@ -746,11 +748,9 @@ public class DeviceServiceImpl implements IDeviceService { valueItem.setId(identity); if (id.startsWith("array_")) { // 查询是否有缓存，如果没有先进行缓存 - if (!redisCache.containsKey(key)) { - Device device = this.selectDeviceBySerialNumber(serialNumber); - this.selectDeviceByDeviceId(device.getDeviceId()); - } - +// if (!redisCache.containsKey(key)) { +// itslValueCache.addCacheDeviceStatus(input.getProductId(), serialNumber); +// } int index = Integer.parseInt(id.substring(6, 8)); if (isShadow) { String[] shadows = cacheIdentifier.getShadow().split(","); @@ -791,7 +791,6 @@ public class DeviceServiceImpl implements IDeviceService { /* ★★★★★★★★★★★★★★★★★★★★★★ 存储数据 - 开始 ★★★★★★★★★★★★★★★★★★★★★★ */ ThingsModelType modelType = ThingsModelType.getType(thingModels.getType()); - Device device = this.selectDeviceBySerialNumber(serialNumber); switch (modelType) { case PROP: if (1 == thingModels.getIsHistory()) { diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/EventLogServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/EventLogServiceImpl.java index 2b8bc211..ebb756e9 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/EventLogServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/EventLogServiceImpl.java @@ -64,4 +64,14 @@ public class EventLogServiceImpl extends ServiceImpl i public List listHistory(EventLog eventLog) { return eventLogMapper.listHistory(eventLog); } + + /** + * 清理事件日志 + * + * @param days 保留天数 + */ + @Override + public void cleanEventLog(int days) { + eventLogMapper.deleteEventLogBeforeDays(days); + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/AlertLogMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/AlertLogMapper.xml index 01d5c50b..ac0ac040 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/AlertLogMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/AlertLogMapper.xml @@ -274,4 +274,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + DELETE FROM iot_alert_log + WHERE create_time < DATE_SUB(NOW(), INTERVAL #{days} DAY) + + diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/EventLogMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/EventLogMapper.xml index f5536dd3..0a1756eb 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/EventLogMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/EventLogMapper.xml @@ -59,4 +59,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" and user_id = #{userId} + + + DELETE FROM iot_event_log + WHERE create_time < DATE_SUB(NOW(), INTERVAL #{days} DAY) + diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml index 783cc576..12a7d254 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml @@ -20,4 +20,14 @@ WHERE sub_serial_number = #{subSerialNumber} + + + diff --git a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysLogininforMapper.java b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysLogininforMapper.java index c68e4946..d9397f81 100644 --- a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysLogininforMapper.java +++ b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysLogininforMapper.java @@ -41,4 +41,12 @@ public interface SysLogininforMapper { public int cleanLogininfor(); public List getUserCount(); + + /** + * 删除指定天数前的登录日志 + * + * @param days 保留天数 + * @return 删除记录数 + */ + public int deleteLogininforBeforeDays(int days); } diff --git a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysOperLogMapper.java b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysOperLogMapper.java index 46ac307e..a05e33b6 100644 --- a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysOperLogMapper.java +++ b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/mapper/SysOperLogMapper.java @@ -45,4 +45,12 @@ public interface SysOperLogMapper { * 清空操作日志 */ public void cleanOperLog(); + + /** + * 删除指定天数前的操作日志 + * + * @param days 保留天数 + * @return 删除记录数 + */ + public int deleteOperLogBeforeDays(int days); } diff --git a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysLogininforService.java b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysLogininforService.java index ef1fcb69..7c27865a 100644 --- a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysLogininforService.java +++ b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysLogininforService.java @@ -42,4 +42,11 @@ public interface ISysLogininforService { * 统计一周内的登陆数据 */ public List getUserCount(); + + /** + * 清理指定天数前的登录日志 + * + * @param days 保留天数 + */ + public void cleanLogininfor(int days); } diff --git a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysOperLogService.java b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysOperLogService.java index b7713433..f2b21e28 100644 --- a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysOperLogService.java +++ b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/ISysOperLogService.java @@ -45,4 +45,11 @@ public interface ISysOperLogService { * 清空操作日志 */ public void cleanOperLog(); + + /** + * 清理指定天数前的操作日志 + * + * @param days 保留天数 + */ + public void cleanOperLog(int days); } diff --git a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysLogininforServiceImpl.java b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysLogininforServiceImpl.java index 3d8cd3d3..e17fdf6f 100644 --- a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysLogininforServiceImpl.java +++ b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysLogininforServiceImpl.java @@ -63,4 +63,14 @@ public class SysLogininforServiceImpl implements ISysLogininforService { public List getUserCount() { return logininforMapper.getUserCount(); } + + /** + * 清理指定天数前的登录日志 + * + * @param days 保留天数 + */ + @Override + public void cleanLogininfor(int days) { + logininforMapper.deleteLogininforBeforeDays(days); + } } diff --git a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysOperLogServiceImpl.java b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysOperLogServiceImpl.java index 5bfed0ee..f0ace9a1 100644 --- a/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysOperLogServiceImpl.java +++ b/fastbee-service/fastbee-system-service/src/main/java/com/fastbee/system/service/impl/SysOperLogServiceImpl.java @@ -68,4 +68,14 @@ public class SysOperLogServiceImpl implements ISysOperLogService { public void cleanOperLog() { operLogMapper.cleanOperLog(); } + + /** + * 清理指定天数前的操作日志 + * + * @param days 保留天数 + */ + @Override + public void cleanOperLog(int days) { + operLogMapper.deleteOperLogBeforeDays(days); + } } diff --git a/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysLogininforMapper.xml b/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysLogininforMapper.xml index 3fe4ccd7..d5ecdf38 100644 --- a/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysLogininforMapper.xml +++ b/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysLogininforMapper.xml @@ -76,4 +76,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" ) b on a.login_time = b.datetime; + + DELETE FROM sys_logininfor + WHERE login_time < DATE_SUB(NOW(), INTERVAL #{days} DAY) + + diff --git a/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysOperLogMapper.xml b/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysOperLogMapper.xml index 38456ed0..13a93f56 100644 --- a/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysOperLogMapper.xml +++ b/fastbee-service/fastbee-system-service/src/main/resources/mapper/system/SysOperLogMapper.xml @@ -80,4 +80,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" truncate table sys_oper_log + + DELETE FROM sys_oper_log + WHERE oper_time < DATE_SUB(NOW(), INTERVAL #{days} DAY) + + \ No newline at end of file diff --git a/sql/log_clean_task_init.sql b/sql/log_clean_task_init.sql new file mode 100644 index 00000000..f12ccdcf --- /dev/null +++ b/sql/log_clean_task_init.sql @@ -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 = '日志清理任务'; \ No newline at end of file diff --git a/yack-web/src/api/platform/deviceMapping.js b/yack-web/src/api/platform/deviceMapping.js index 380ee19f..920f19e4 100644 --- a/yack-web/src/api/platform/deviceMapping.js +++ b/yack-web/src/api/platform/deviceMapping.js @@ -28,6 +28,11 @@ export function listSubMappings(parentId) { 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) { return request({ url: `/platform/device-mapping/${parentId}/sub-mappings`, method: 'post', data }) } diff --git a/yack-web/src/views/platform/device-mapping/index.vue b/yack-web/src/views/platform/device-mapping/index.vue index 0b5f70e8..2a4aaa49 100644 --- a/yack-web/src/views/platform/device-mapping/index.vue +++ b/yack-web/src/views/platform/device-mapping/index.vue @@ -335,7 +335,7 @@