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()); + *+ * + *
工作原理: + *
说明: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 核心特性:
+ * 使用示例:
+ * 格式:基于时间戳 + 序列号 + 随机数
+ * 示例:752431931275776594
+ *
+ * 特点:
+ * 位域分配(低到高):
+ * 注意: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', -- 禁止并发执行
+ *
+ *
+ *
+ * // 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
+ *
+ *
+ * @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