From d7a524cba34ecab8e8212806424e3d95a982849c Mon Sep 17 00:00:00 2001 From: zhugongping Date: Tue, 15 Sep 2026 15:25:50 +0800 Subject: [PATCH] =?UTF-8?q?20260915:traceId=E6=97=A5=E5=BF=97=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E8=BF=BD=E8=B8=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../knowledge-base/07-TraceId 链路追踪实现.md | 1213 +++++++++++++++++ .../com/fastbee/common/config/DeviceTask.java | 13 +- .../common/constant/FastBeeConstant.java | 2 + .../fastbee/common/core/mq/DeviceReplyBo.java | 3 + .../common/core/mq/DeviceReportBo.java | 3 + .../common/core/mq/DeviceStatusBo.java | 4 + .../common/core/mq/DeviceTestReportBo.java | 3 + .../fastbee/common/core/mq/InvokeReqDto.java | 3 + .../common/core/mq/MQSendMessageBo.java | 3 + .../common/core/mq/ota/OtaUpgradeBo.java | 3 + .../core/mq/ota/OtaUpgradeDelayTask.java | 5 + .../common/trace/MdcTaskDecorator.java | 75 + .../fastbee/common/trace/TraceContext.java | 270 ++++ .../common/trace/TraceIdGenerator.java | 106 ++ .../fastbee/common/utils/TraceIdUtils.java | 184 +-- .../aspectj/ScheduledTraceAspect.java | 47 + .../framework/config/OtaThreadPoolConfig.java | 5 + .../framework/config/ThreadPoolConfig.java | 5 + .../interceptor/TraceIdInterceptor.java | 22 +- .../iot/data/job/DeviceAbstractQuartzJob.java | 13 +- .../iot/data/listen/DeviceOtherListen.java | 11 +- .../data/listen/DevicePropFetchListen.java | 14 +- .../iot/data/listen/DeviceReplyListen.java | 13 +- .../iot/data/listen/DeviceReportListen.java | 16 +- .../iot/data/listen/DeviceStatusListen.java | 13 +- .../iot/data/listen/DeviceTestListen.java | 10 +- .../iot/data/listen/FunctionInvokeListen.java | 11 +- .../iot/data/listen/UpGradeListener.java | 15 +- .../iot/data/listen/UpgradeListen.java | 15 +- .../service/impl/FirmwareTaskServiceImpl.java | 5 +- .../fastbee/mq/producer/MessageProducer.java | 45 + .../quartz/util/AbstractQuartzJob.java | 11 +- .../record/config/ThreadPoolTaskConfig.java | 5 + .../base/core/hanler/TraceHandler.java | 38 + .../java/com/fastbee/server/TCPServer.java | 5 +- .../java/com/fastbee/server/UDPServer.java | 3 + .../server/handler/DispatcherHandler.java | 13 +- .../handler/adapter/MqttMessageDelegate.java | 2 + .../com/fastbee/mqtt/server/MqttServer.java | 3 + .../fastbee/mqtt/server/WebSocketServer.java | 3 + .../iot/model/modbus/ModbusPollJob.java | 3 + sql/log_clean_task_init.sql | 2 +- 42 files changed, 2044 insertions(+), 194 deletions(-) create mode 100644 docs/knowledge-base/07-TraceId 链路追踪实现.md create mode 100644 fastbee-common/src/main/java/com/fastbee/common/trace/MdcTaskDecorator.java create mode 100644 fastbee-common/src/main/java/com/fastbee/common/trace/TraceContext.java create mode 100644 fastbee-common/src/main/java/com/fastbee/common/trace/TraceIdGenerator.java create mode 100644 fastbee-framework/src/main/java/com/fastbee/framework/aspectj/ScheduledTraceAspect.java create mode 100644 fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/TraceHandler.java 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-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. + *
  3. 在任务执行时,恢复快照到新线程(MDC 由 TraceContext 统一同步)
  4. + *
  5. 在任务完成后,清理上下文防止线程池污染
  6. + *
+ * + *

说明: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/TraceIdUtils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java index fdd7138c..558a4223 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 + * @deprecated 建议直接使用 {@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-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-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..356168fb 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; @@ -56,8 +57,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()); } @@ -106,10 +107,10 @@ public abstract class DeviceAbstractQuartzJob implements Job { // 写入数据库当中 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/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..8f3cf966 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,8 +28,16 @@ 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); } 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..072351c0 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,7 +27,14 @@ 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数据转发异常"); } 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..8f24f30e 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,7 +27,15 @@ 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); } 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/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放置在协议 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-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/sql/log_clean_task_init.sql b/sql/log_clean_task_init.sql index 36e0efb2..f12ccdcf 100644 --- a/sql/log_clean_task_init.sql +++ b/sql/log_clean_task_init.sql @@ -19,7 +19,7 @@ INSERT INTO `sys_job` ( ) VALUES ( '日志清理任务', -- 任务名称 'SYSTEM', -- 任务组名 - 'logCleanTask.cleanLogs()', -- 调用目标字符串 + 'logCleanTask.cleanLogs(30)', -- 调用目标字符串 '0 0 0 * * ?', -- Cron 表达式:每天凌晨 00:00:00 '3', -- 错误策略:不触发立即执行 '1', -- 禁止并发执行