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 1bf001da..be51c4db 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,7 +1,6 @@ package com.fastbee.common.config; import com.fastbee.common.constant.FastBeeConstant; -import com.fastbee.common.utils.TraceIdUtils; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -106,7 +105,6 @@ public class DeviceTask { executor.setMaxPoolSize(maxSize); executor.setKeepAliveSeconds(keepAlive); executor.setQueueCapacity(queueCapacity); - executor.setTaskDecorator(TraceIdUtils::wrapWithMdc); // 线程池对拒绝任务的处理策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); //线程池名的前缀 diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/config/MdcTaskDecorator.java b/fastbee-framework/src/main/java/com/fastbee/framework/config/MdcTaskDecorator.java deleted file mode 100644 index 1f0488b7..00000000 --- a/fastbee-framework/src/main/java/com/fastbee/framework/config/MdcTaskDecorator.java +++ /dev/null @@ -1,18 +0,0 @@ -package com.fastbee.framework.config; - -import com.fastbee.common.utils.TraceIdUtils; -import org.springframework.core.task.TaskDecorator; - -/** - * 线程池 MDC 透传装饰器： - * - 透传提交线程 MDC - * - 兜底补齐 traceId - * - 执行结束后还原/清理，避免线程复用污染 - */ -public class MdcTaskDecorator implements TaskDecorator { - @Override - public Runnable decorate(Runnable runnable) { - return TraceIdUtils.wrapWithMdc(runnable); - } -} - diff --git a/fastbee-framework/src/main/java/com/fastbee/framework/config/SchedulingConfig.java b/fastbee-framework/src/main/java/com/fastbee/framework/config/SchedulingConfig.java deleted file mode 100644 index 95fd8892..00000000 --- a/fastbee-framework/src/main/java/com/fastbee/framework/config/SchedulingConfig.java +++ /dev/null @@ -1,27 +0,0 @@ -package com.fastbee.framework.config; - -import org.springframework.context.annotation.Configuration; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.scheduling.annotation.SchedulingConfigurer; -import org.springframework.scheduling.config.ScheduledTaskRegistrar; - -import javax.annotation.Resource; -import java.util.concurrent.ScheduledExecutorService; - -/** - * 统一 @Scheduled 执行线程池，确保： - * - 使用项目自定义的 scheduledExecutorService（已做 MDC 透传与异常打印） - */ -@Configuration -@EnableScheduling -public class SchedulingConfig implements SchedulingConfigurer { - - @Resource(name = "scheduledExecutorService") - private ScheduledExecutorService scheduledExecutorService; - - @Override - public void configureTasks(ScheduledTaskRegistrar taskRegistrar) { - taskRegistrar.setScheduler(scheduledExecutorService); - } -} - 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 90615839..6b25dd4c 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,16 +1,13 @@ package com.fastbee.framework.config; import com.fastbee.common.utils.Threads; -import com.fastbee.common.utils.TraceIdUtils; import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import java.util.concurrent.Callable; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.RunnableScheduledFuture; import java.util.concurrent.ThreadPoolExecutor; /** @@ -39,10 +36,8 @@ public class ThreadPoolConfig { executor.setCorePoolSize(corePoolSize); executor.setQueueCapacity(queueCapacity); executor.setKeepAliveSeconds(keepAliveSeconds); - executor.setTaskDecorator(new MdcTaskDecorator()); // 线程池对拒绝任务(无线程可用)的处理策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); - executor.initialize(); return executor; } @@ -54,17 +49,6 @@ public class ThreadPoolConfig { return new ScheduledThreadPoolExecutor(corePoolSize, new BasicThreadFactory.Builder().namingPattern("schedule-pool-%d").daemon(true).build(), new ThreadPoolExecutor.CallerRunsPolicy()) { - - @Override - protected RunnableScheduledFuture decorateTask(Runnable runnable, RunnableScheduledFuture task) { - return super.decorateTask(TraceIdUtils.wrapWithMdc(runnable), task); - } - - @Override - protected RunnableScheduledFuture decorateTask(Callable callable, RunnableScheduledFuture task) { - return super.decorateTask(TraceIdUtils.wrapWithMdc(callable), task); - } - @Override protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r, t); 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 31099277..f352620a 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceReportBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceOtherMsgConsumer; import com.fastbee.mq.queue.DeviceOtherQueue; import lombok.extern.slf4j.Slf4j; @@ -27,13 +26,9 @@ public class DeviceOtherListen { while (true) { try { DeviceReportBo reportBo = DeviceOtherQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); otherMsgConsumer.consume(reportBo); } catch (Exception e) { log.error("=>emq数据转发异常"); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 3e65fc82..98625699 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 @@ -4,7 +4,6 @@ import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.iot.data.consumer.DevicePropFetchConsumer; import com.fastbee.iot.model.modbus.ModbusPollJob; import com.fastbee.mq.queue.DevicePropFetchQueue; -import com.fastbee.common.utils.TraceIdUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; @@ -28,14 +27,9 @@ public class DevicePropFetchListen { while (true) { try { ModbusPollJob take = DevicePropFetchQueue.take(); - // 队列消费是典型的异步边界：每条消息独立生成 traceId，避免线程复用导致 traceId 串线 - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); devicePropFetchConsumer.consume(take); } catch (Exception e) { log.error("=>设备属性获取异常", e); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 7d34bda0..338f6012 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceReportBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceReplyMsgConsumer; import com.fastbee.mq.queue.DeviceReplyQueue; import lombok.extern.slf4j.Slf4j; @@ -28,14 +27,10 @@ public class DeviceReplyListen { try { /*读队列消息*/ DeviceReportBo reportBo = DeviceReplyQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); /*处理消息*/ deviceReplyMsgHandler.consume(reportBo); } catch (Exception e) { log.error("=>设备回调消息监听异常", e); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 8561992a..3276d99a 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceReportBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceReportMsgConsumer; import com.fastbee.mq.queue.DeviceReportQueue; import lombok.extern.slf4j.Slf4j; @@ -28,14 +27,10 @@ public class DeviceReportListen { try { /*取出数据*/ DeviceReportBo reportBo = DeviceReportQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); /*处理数据*/ reportMsgConsumer.consume(reportBo); } catch (Exception e) { log.error("=>设备上报数据监听异常", e); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 ad3f8351..d67f8239 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceStatusBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceStatusConsumer; import com.fastbee.mq.queue.DeviceStatusQueue; import lombok.extern.slf4j.Slf4j; @@ -24,17 +23,13 @@ public class DeviceStatusListen { @Async(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK) public void listen() { - while (true) { - try { + try { + while (true) { DeviceStatusBo status = DeviceStatusQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); deviceStatusConsumer.consume(status); - } catch (Exception e) { - log.error("设备状态监听错误", e); - } finally { - TraceIdUtils.clearTraceId(); } + } catch (Exception e) { + log.error("设备状态监听错误", e); } } 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 b1573e74..91147170 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.DeviceTestReportBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceTestConsumer; import com.fastbee.mq.queue.DeviceTestQueue; import lombok.extern.slf4j.Slf4j; @@ -26,14 +25,10 @@ public class DeviceTestListen { while (true) { try { DeviceTestReportBo take = DeviceTestQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); deviceTestConsumer.consume(take); } catch (Exception e) { log.error("=>emq数据转发异常"); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 2c85d83c..d7f18f47 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.MQSendMessageBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.FunctionInvokeConsumer; import com.fastbee.mq.queue.FunctionInvokeQueue; import lombok.extern.slf4j.Slf4j; @@ -27,13 +26,9 @@ public class FunctionInvokeListen { while (true) { try { MQSendMessageBo sendBo = FunctionInvokeQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); functionInvokeConsumer.handler(sendBo); } catch (Exception e) { log.error("=>下发服务消费异常", e); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 2dc3457d..e0392944 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 @@ -5,7 +5,6 @@ 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.utils.StringUtils; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.service.IOtaUpgradeService; import com.fastbee.mq.queue.DelayUpgradeQueue; import com.fastbee.system.service.ISysMenuService; @@ -48,8 +47,6 @@ public class UpGradeListener implements ApplicationRunner { try { OtaUpgradeDelayTask task = DelayUpgradeQueue.task(); if (StringUtils.isNotNull(task)) { - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); Date startTime = task.getStartTime(); otaUpgradeService.upgrade(task); log.info("=>开始OTA升级时间{}", startTime); @@ -57,8 +54,6 @@ public class UpGradeListener implements ApplicationRunner { continue; } catch (Exception e) { log.error("=>OTA升级监听异常", e); - } finally { - TraceIdUtils.clearTraceId(); } } } 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 1255d91e..0bc713ab 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,7 +2,6 @@ package com.fastbee.iot.data.listen; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.mq.ota.OtaUpgradeBo; -import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.mq.queue.OtaUpgradeQueue; import lombok.extern.slf4j.Slf4j; @@ -29,14 +28,10 @@ public class UpgradeListen { try { /*获取队列中的OTA升级消息*/ OtaUpgradeBo upgradeBo = OtaUpgradeQueue.take(); - TraceIdUtils.clearTraceId(); - TraceIdUtils.ensureTraceId(); // OTA升级处理 functionSendService.upGradeOTA(upgradeBo); } catch (Exception e) { log.error("->OTA消息监听异常", e); - } finally { - TraceIdUtils.clearTraceId(); } } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java index d461ad02..718e6dd6 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java @@ -92,8 +92,7 @@ public class SceneContext { * @return */ private boolean process(String json) throws InterruptedException { - // 场景执行很多时候不在 Web 请求线程内，这里兜底补齐 traceId - TraceIdUtils.ensureTraceId(); + ScriptTemplate scriptTemplate = JSON.parseObject(json, ScriptTemplate.class); String requestId = "scene/" + scriptTemplate.getChainName(); scene_logger.info("[{}]=====+>场景联动规则开始执行！", requestId); diff --git a/fastbee-record/src/main/java/com/fastbee/record/config/SchedulingConfig.java b/fastbee-record/src/main/java/com/fastbee/record/config/SchedulingConfig.java deleted file mode 100644 index f2256623..00000000 --- a/fastbee-record/src/main/java/com/fastbee/record/config/SchedulingConfig.java +++ /dev/null @@ -1,46 +0,0 @@ -package com.fastbee.record.config; - -import com.fastbee.common.utils.TraceIdUtils; -import org.apache.commons.lang3.concurrent.BasicThreadFactory; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.scheduling.annotation.SchedulingConfigurer; -import org.springframework.scheduling.config.ScheduledTaskRegistrar; - -import java.util.concurrent.Callable; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.RunnableScheduledFuture; -import java.util.concurrent.ThreadPoolExecutor; - -/** - * record 模块自身的 @Scheduled 配置（该模块 ComponentScan 不包含 fastbee-framework）。 - * 统一使用带 MDC 透传的线程池执行定时任务。 - */ -@Configuration -public class SchedulingConfig implements SchedulingConfigurer { - - @Bean(name = "recordScheduledExecutorService") - public ScheduledExecutorService recordScheduledExecutorService() { - int corePoolSize = Math.max(2, Runtime.getRuntime().availableProcessors()); - return new ScheduledThreadPoolExecutor(corePoolSize, - new BasicThreadFactory.Builder().namingPattern("record-schedule-pool-%d").daemon(true).build(), - new ThreadPoolExecutor.CallerRunsPolicy()) { - - @Override - protected RunnableScheduledFuture decorateTask(Runnable runnable, RunnableScheduledFuture task) { - return super.decorateTask(TraceIdUtils.wrapWithMdc(runnable), task); - } - - @Override - protected RunnableScheduledFuture decorateTask(Callable callable, RunnableScheduledFuture task) { - return super.decorateTask(TraceIdUtils.wrapWithMdc(callable), task); - } - }; - } - - @Override - public void configureTasks(ScheduledTaskRegistrar taskRegistrar) { - taskRegistrar.setScheduler(recordScheduledExecutorService()); - } -} diff --git a/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java b/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java index 4bca78af..703e36c2 100644 --- a/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java +++ b/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java @@ -1,6 +1,5 @@ package com.fastbee.record.config; -import com.fastbee.common.utils.TraceIdUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @@ -49,7 +48,6 @@ public class ThreadPoolTaskConfig { executor.setQueueCapacity(queueCapacity); executor.setKeepAliveSeconds(keepAliveTime); executor.setThreadNamePrefix(threadNamePrefix); - executor.setTaskDecorator(TraceIdUtils::wrapWithMdc); // 线程池对拒绝任务的处理策略 // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务 diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java index e0bc7837..ac8d78ab 100644 --- a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java @@ -1,6 +1,5 @@ package com.fastbee.sip.conf; -import com.fastbee.common.utils.TraceIdUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @@ -26,7 +25,6 @@ public class ThreadPoolTaskConfig { executor.setQueueCapacity(queueCapacity); executor.setKeepAliveSeconds(keepAliveTime); executor.setThreadNamePrefix(threadNamePrefix); - executor.setTaskDecorator(TraceIdUtils::wrapWithMdc); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.initialize(); return executor;