diff --git a/fastbee-admin/src/main/resources/logback.xml b/fastbee-admin/src/main/resources/logback.xml index 3489f2bb..90f5f198 100644 --- a/fastbee-admin/src/main/resources/logback.xml +++ b/fastbee-admin/src/main/resources/logback.xml @@ -107,7 +107,7 @@ 3 - %d{HH:mm:ss.SSS} [%method,%line] - %msg%n + %d{HH:mm:ss.SSS} [%thread] [%X{traceId}] [%method,%line] - %msg%n INFO @@ -123,7 +123,7 @@ 3 - %d{HH:mm:ss.SSS} [%method,%line] - %msg%n + %d{HH:mm:ss.SSS} [%thread] [%X{traceId}] [%method,%line] - %msg%n INFO 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 be51c4db..1bf001da 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.utils.TraceIdUtils; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -105,6 +106,7 @@ 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-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java index a6dfe2d2..bb57d2ad 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 @@ -3,6 +3,8 @@ package com.fastbee.common.utils; import com.fastbee.common.utils.uuid.IdUtils; import org.slf4j.MDC; +import java.util.Map; +import java.util.concurrent.Callable; import java.util.concurrent.ThreadLocalRandom; /** @@ -101,6 +103,81 @@ public class TraceIdUtils { 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 则补齐;执行完自动清理/还原。 + */ + public static Runnable wrapWithMdc(Runnable task) { + final Map captured = MDC.getCopyOfContextMap(); + return () -> { + Map previous = MDC.getCopyOfContextMap(); + 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); + } + } + }; + } + + /** + * 包装 Callable:透传提交线程的 MDC,上下文中无 traceId 则补齐;执行完自动清理/还原。 + */ + public static Callable wrapWithMdc(Callable task) { + final Map captured = MDC.getCopyOfContextMap(); + return () -> { + Map previous = MDC.getCopyOfContextMap(); + 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); + } + } + }; + } + /** * 清除MDC中的TraceId */ 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 new file mode 100644 index 00000000..1f0488b7 --- /dev/null +++ b/fastbee-framework/src/main/java/com/fastbee/framework/config/MdcTaskDecorator.java @@ -0,0 +1,18 @@ +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 new file mode 100644 index 00000000..cb1183b6 --- /dev/null +++ b/fastbee-framework/src/main/java/com/fastbee/framework/config/SchedulingConfig.java @@ -0,0 +1,25 @@ +package com.fastbee.framework.config; + +import org.springframework.context.annotation.Configuration; +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 +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 6b25dd4c..90615839 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,13 +1,16 @@ 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; /** @@ -36,8 +39,10 @@ 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; } @@ -49,6 +54,17 @@ 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/consumer/DeviceStatusConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java index b465291d..81b5e437 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java @@ -7,14 +7,19 @@ import com.fastbee.common.core.mq.DeviceStatusBo; import com.fastbee.common.core.mq.MQSendMessageBo; import com.fastbee.common.core.mq.SubDeviceBo; import com.fastbee.common.core.mq.message.ReportDataBo; +import com.fastbee.common.core.mq.message.DeviceMessage; import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem; import com.fastbee.common.enums.DeviceStatus; +import com.fastbee.common.core.redis.RedisCache; +import com.fastbee.common.utils.StringUtils; +import com.fastbee.common.enums.TopicType; import com.fastbee.iot.cache.IDeviceCache; import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.iot.data.service.IRuleEngine; import com.fastbee.iot.domain.Device; import com.fastbee.iot.domain.Product; import com.fastbee.iot.model.ThingsModels.ThingsModelShadow; +import com.fastbee.iot.data.service.IDeviceMessageService; import com.fastbee.iot.service.IDeviceService; import com.fastbee.iot.service.IProductService; import com.fastbee.iot.service.ISubGatewayService; @@ -30,6 +35,7 @@ import org.springframework.util.CollectionUtils; import javax.annotation.Resource; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; /** * 设备状态消息处理类 @@ -57,8 +63,20 @@ public class DeviceStatusConsumer { private Boolean enabled; @Resource private ISubGatewayService subGatewayService; + + @Resource + private IDeviceMessageService deviceMessageService; + + @Resource + private RedisCache redisCache; + private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(2); + private static final String IMEI_CMD = "AT+IMEI\\r\\n"; + private static final String ICCID_CMD = "AT+ICCID\\r\\n"; + private static final int AUTO_FETCH_COOLDOWN_SECONDS = 300; + private static final String AUTO_FETCH_KEY_PREFIX = "iot:autoFetch:imeiIccid:"; + public synchronized void consume(DeviceStatusBo bo) { try { List list = new ArrayList<>(); @@ -90,6 +108,10 @@ public class DeviceStatusConsumer { } /*更新设备状态*/ deviceCache.updateDeviceStatusCache(statusBo, device); + // 设备首次上线,自动下发 AT 指令获取 IMEI/ICCID(字段为空才下发,并做简单限流) + if (status.equals(DeviceStatus.ONLINE)) { + autoFetchImeiIccidIfNeeded(device); + } //处理影子模式值 this.handlerShadow(device, status); //设备上下线执行规则引擎 @@ -107,6 +129,45 @@ public class DeviceStatusConsumer { } } + private void autoFetchImeiIccidIfNeeded(Device device) { + if (device == null || StringUtils.isEmpty(device.getSerialNumber())) { + return; + } + String serialNumber = device.getSerialNumber(); + + boolean needImei = StringUtils.isEmpty(device.getImei()); + boolean needIccid = StringUtils.isEmpty(device.getIccid()); + if (!needImei && !needIccid) { + return; + } + + if (needImei) { + String key = AUTO_FETCH_KEY_PREFIX + serialNumber + ":imei"; + if (!redisCache.containsKey(key)) { + redisCache.setCacheObject(key, "1", AUTO_FETCH_COOLDOWN_SECONDS, TimeUnit.SECONDS); + sendAtCommand(serialNumber, IMEI_CMD); + } + } + + if (needIccid) { + String key = AUTO_FETCH_KEY_PREFIX + serialNumber + ":iccid"; + if (!redisCache.containsKey(key)) { + redisCache.setCacheObject(key, "1", AUTO_FETCH_COOLDOWN_SECONDS, TimeUnit.SECONDS); + sendAtCommand(serialNumber, ICCID_CMD); + } + } + } + + private void sendAtCommand(String serialNumber, String command) { + DeviceMessage msg = new DeviceMessage(); + msg.setSerialNumber(serialNumber); + msg.setTopicName(TopicType.FUNCTION_GET.getTopicSuffix()); + msg.setMessage(command); + msg.setDataType("ascii"); + deviceMessageService.messagePost(msg); + log.info("自动下发 AT 指令获取信息, serialNumber={}, cmd={}", serialNumber, command); + } + private void handlerShadow(Device device, DeviceStatus status) { //获取设备协议编码 DeviceAndProtocol dp = deviceService.selectProtocolBySerialNumber(device.getSerialNumber()); 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 f352620a..31099277 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.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceOtherMsgConsumer; import com.fastbee.mq.queue.DeviceOtherQueue; import lombok.extern.slf4j.Slf4j; @@ -26,9 +27,13 @@ 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 98625699..3e65fc82 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,6 +4,7 @@ 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; @@ -27,9 +28,14 @@ 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 338f6012..7d34bda0 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.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceReplyMsgConsumer; import com.fastbee.mq.queue.DeviceReplyQueue; import lombok.extern.slf4j.Slf4j; @@ -27,10 +28,14 @@ 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 3276d99a..8561992a 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.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceReportMsgConsumer; import com.fastbee.mq.queue.DeviceReportQueue; import lombok.extern.slf4j.Slf4j; @@ -27,10 +28,14 @@ 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 d67f8239..ad3f8351 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.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceStatusConsumer; import com.fastbee.mq.queue.DeviceStatusQueue; import lombok.extern.slf4j.Slf4j; @@ -23,13 +24,17 @@ public class DeviceStatusListen { @Async(FastBeeConstant.TASK.MESSAGE_CONSUME_TASK) public void listen() { - try { - while (true) { + while (true) { + try { 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 91147170..b1573e74 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.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.DeviceTestConsumer; import com.fastbee.mq.queue.DeviceTestQueue; import lombok.extern.slf4j.Slf4j; @@ -25,10 +26,14 @@ 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 d7f18f47..2c85d83c 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.utils.TraceIdUtils; import com.fastbee.iot.data.consumer.FunctionInvokeConsumer; import com.fastbee.mq.queue.FunctionInvokeQueue; import lombok.extern.slf4j.Slf4j; @@ -26,9 +27,13 @@ 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 e0392944..2dc3457d 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,6 +5,7 @@ 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; @@ -47,6 +48,8 @@ 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); @@ -54,6 +57,8 @@ 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 0bc713ab..1255d91e 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.utils.TraceIdUtils; import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.mq.queue.OtaUpgradeQueue; import lombok.extern.slf4j.Slf4j; @@ -28,10 +29,14 @@ 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 591fcc52..0e39daf2 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 @@ -12,6 +12,7 @@ import com.fastbee.common.exception.ServiceException; import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.SecurityUtils; import com.fastbee.common.utils.StringUtils; +import com.fastbee.common.utils.TraceIdUtils; import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.iot.data.service.IFunctionInvoke; import com.fastbee.iot.data.service.IRuleEngine; @@ -91,11 +92,12 @@ 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); - scene_logger.info("[{}]=====+>规则引擎脚本配置:{},规则引擎上下文:{}", requestId, scriptTemplate, this); + scene_logger.info("[{}]=====+>规则引擎脚本配置:{},规则引擎上下文:{}", requestId, JSON.toJSONString(scriptTemplate), JSON.toJSONString(this)); // MDC.put("id", scriptTemplate.getChainName()); // 触发器 if (scriptTemplate.getPurpose() == 2) { @@ -377,7 +379,7 @@ public class SceneContext { // 默认上线执行告警恢复 在延迟匹配周期内不告警 ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); int delayTime = scriptTemplate.getCheckdelay() > 0 ? scriptTemplate.getCheckdelay() : scriptTemplate.getDelay(); - executor.schedule(() -> { + executor.schedule(TraceIdUtils.wrapWithMdc(() -> { // 设备未上线,则执行告警任务 if (scriptTemplate.getSource() == 4) { // 告警 @@ -386,7 +388,7 @@ public class SceneContext { // 下发指令 this.send(scriptTemplate); } - }, delayTime, TimeUnit.SECONDS); + }), delayTime, TimeUnit.SECONDS); scene_logger.info("[{}]=====+>延时执行告警任务:{}", requestId, key); ruleService.addCheckTask(key, executor); } 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 new file mode 100644 index 00000000..f2256623 --- /dev/null +++ b/fastbee-record/src/main/java/com/fastbee/record/config/SchedulingConfig.java @@ -0,0 +1,46 @@ +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 703e36c2..4bca78af 100644 --- a/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java +++ b/fastbee-record/src/main/java/com/fastbee/record/config/ThreadPoolTaskConfig.java @@ -1,5 +1,6 @@ package com.fastbee.record.config; +import com.fastbee.common.utils.TraceIdUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @@ -48,6 +49,7 @@ 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 ac8d78ab..e0bc7837 100644 --- a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/conf/ThreadPoolTaskConfig.java @@ -1,5 +1,6 @@ package com.fastbee.sip.conf; +import com.fastbee.common.utils.TraceIdUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @@ -25,6 +26,7 @@ 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; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/Device.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/Device.java index dcdb8045..58b8e386 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/Device.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/Device.java @@ -227,6 +227,18 @@ public class Device extends BaseEntity { @TableField(exist = false) private String transport; + /** + * IMEI(蜂窝模块设备标识) + */ + @ApiModelProperty("IMEI") + private String imei; + + /** + * ICCID(SIM卡标识) + */ + @ApiModelProperty("ICCID") + private String iccid; + @Transient @ApiModelProperty("机构id") @TableField(exist = false) diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceAllShortOutput.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceAllShortOutput.java index a76d7c72..44a1ecb2 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceAllShortOutput.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceAllShortOutput.java @@ -44,6 +44,16 @@ public class DeviceAllShortOutput { */ private String serialNumber; + /** + * IMEI(蜂窝模块设备标识) + */ + private String imei; + + /** + * ICCID(SIM卡标识) + */ + private String iccid; + /** * 固件版本 */ @@ -217,6 +227,22 @@ public class DeviceAllShortOutput { this.serialNumber = serialNumber; } + public String getImei() { + return imei; + } + + public void setImei(String imei) { + this.imei = imei; + } + + public String getIccid() { + return iccid; + } + + public void setIccid(String iccid) { + this.iccid = iccid; + } + public BigDecimal getFirmwareVersion() { return firmwareVersion; } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceShortOutput.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceShortOutput.java index 0578c3fd..2fd4cb8e 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceShortOutput.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceShortOutput.java @@ -56,6 +56,16 @@ public class DeviceShortOutput { */ @Excel(name = "设备编号") private String serialNumber; + + /** + * IMEI(蜂窝模块设备标识) + */ + private String imei; + + /** + * ICCID(SIM卡标识) + */ + private String iccid; /** * 固件版本 */ @@ -396,6 +406,22 @@ public class DeviceShortOutput { this.serialNumber = serialNumber; } + public String getImei() { + return imei; + } + + public void setImei(String imei) { + this.imei = imei; + } + + public String getIccid() { + return iccid; + } + + public void setIccid(String iccid) { + this.iccid = iccid; + } + public BigDecimal getFirmwareVersion() { return firmwareVersion; } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java index a7cf5c15..13beda65 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java @@ -303,7 +303,7 @@ public class DeviceServiceImpl implements IDeviceService { } device.setUpdateTime(DateUtils.getNowDate()); // 未激活状态,可以修改产品以及物模型值 - if (device.getStatus() == 1) { + if (Integer.valueOf(1).equals(device.getStatus())) { // 缓存设备状态(物模型值) itslValueCache.addCacheDeviceStatus(device.getProductId(), device.getSerialNumber()); diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceMapper.xml index 3cc25b70..4f137fbc 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceMapper.xml @@ -36,6 +36,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + @@ -64,6 +66,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + @@ -84,6 +88,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + @@ -187,6 +193,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" update_time, img_url, summary, + imei, + iccid, remark, slave_id from iot_device @@ -208,6 +216,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" location_way, things_model_value, active_time, + imei, + iccid, img_url, slave_id from iot_device @@ -434,7 +444,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" select d.device_id, d.device_name, d.product_id, p.product_name,p.device_type, d.tenant_id, d.tenant_name, d.serial_number, d.firmware_version,d.wireless_version, d.status, d.rssi,d.is_shadow,d.is_simulate ,d.location_way,d.things_model_value, d.network_address, d.network_ip, d.longitude, d.latitude, d.active_time, d.create_time, d.update_time, - d.img_url,d.summary,d.remark,p.guid, p.transport, p.protocol_code from iot_device d + d.img_url,d.summary,d.imei,d.iccid,d.remark,p.guid, p.transport, p.protocol_code from iot_device d left join iot_product p on p.product_id=d.product_id where device_id = #{deviceId} @@ -657,6 +667,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" remark = #{remark}, img_url = #{imgUrl}, summary = #{summary}, + imei = #{imei}, + iccid = #{iccid}, slave_id = #{slaveId}, where device_id = #{deviceId} @@ -718,6 +730,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" img_url = #{imgUrl}, summary = #{summary}, gw_dev_code = #{gwDevCode}, + imei = #{imei}, + iccid = #{iccid}, where serial_number = #{serialNumber} diff --git a/yack-web/src/views/iot/device/running-status.vue b/yack-web/src/views/iot/device/running-status.vue index 5875c3d8..10802cfd 100644 --- a/yack-web/src/views/iot/device/running-status.vue +++ b/yack-web/src/views/iot/device/running-status.vue @@ -21,6 +21,39 @@ {{ $t('device.running-status.866086-2') }} + + + +
+ + +
编辑下发指令
+ + +
+ 从设备读取 +
+
+ + + +
+ + +
编辑下发指令
+ + +
+ 从设备读取 +
+
+