Sumuxi/fastbee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260329:日志打印TradeId

Browse Source
This commit is contained in:
George 2026-03-29 13:31:05 +08:00
parent acc9768619
commit 3f3d14f492
17 changed files with 5 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java
View File

@ -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());
//线程池名的前缀

18
fastbee-framework/src/main/java/com/fastbee/framework/config/MdcTaskDecorator.java
View File

@ -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);
}
}

27
fastbee-framework/src/main/java/com/fastbee/framework/config/SchedulingConfig.java
View File

@ -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);
}
}

16
fastbee-framework/src/main/java/com/fastbee/framework/config/ThreadPoolConfig.java
View File

@ -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 <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
return super.decorateTask(TraceIdUtils.wrapWithMdc(runnable), task);
}
@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
return super.decorateTask(TraceIdUtils.wrapWithMdc(callable), task);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceOtherListen.java
View File

@ -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();
}
}
}

6
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DevicePropFetchListen.java
View File

@ -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();
}
}
}

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReplyListen.java
View File

@ -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();
}
}
}

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceReportListen.java
View File

@ -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();
}
}
}

9
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceStatusListen.java
View File

@ -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 {
while (true) {
DeviceStatusBo status = DeviceStatusQueue.take();
TraceIdUtils.clearTraceId();
TraceIdUtils.ensureTraceId();
deviceStatusConsumer.consume(status);
}
} catch (Exception e) {
log.error("设备状态监听错误", e);
} finally {
TraceIdUtils.clearTraceId();
}
}
}

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/DeviceTestListen.java
View File

@ -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();
}
}
}

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/FunctionInvokeListen.java
View File

@ -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();
}
}
}

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpGradeListener.java
View File

@ -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();
}
}
}

5
fastbee-iot-data/src/main/java/com/fastbee/iot/data/listen/UpgradeListen.java
View File

@ -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();
}
}
}

3
fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java
View File

@ -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);

46
fastbee-record/src/main/java/com/fastbee/record/config/SchedulingConfig.java
View File

@ -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 <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
return super.decorateTask(TraceIdUtils.wrapWithMdc(runnable), task);
}
@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
return super.decorateTask(TraceIdUtils.wrapWithMdc(callable), task);
}
};
}
@Override
public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
taskRegistrar.setScheduler(recordScheduledExecutorService());
}
}

2
fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java
View File

@ -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：由调用线程（提交任务的线程）处理该任务

2
fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java
View File

@ -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;