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

20260319:1调整线程traceId;2获取ICCID/IMEI（暂存版）

Browse Source
This commit is contained in:
George 2026-03-19 08:06:59 +08:00
parent 90b078323b
commit 681cd4a9db
26 changed files with 519 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
fastbee-admin/src/main/resources/logback.xml
View File

@ -107,7 +107,7 @@
<maxHistory>3</maxHistory>
</rollingPolicy>
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%method,%line] - %msg%n</pattern>
<pattern>%d{HH:mm:ss.SSS} [%thread] [%X{traceId}] [%method,%line] - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
@ -123,7 +123,7 @@
<maxHistory>3</maxHistory>
</rollingPolicy>
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%method,%line] - %msg%n</pattern>
<pattern>%d{HH:mm:ss.SSS} [%thread] [%X{traceId}] [%method,%line] - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>

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

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

77
fastbee-common/src/main/java/com/fastbee/common/utils/TraceIdUtils.java
View File

@ -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<String, String> captured = MDC.getCopyOfContextMap();
return () -> {
Map<String, String> 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 <T> Callable<T> wrapWithMdc(Callable<T> task) {
final Map<String, String> captured = MDC.getCopyOfContextMap();
return () -> {
Map<String, String> 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
*/

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

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

25
fastbee-framework/src/main/java/com/fastbee/framework/config/SchedulingConfig.java Normal file
View File

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

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

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

61
fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java
View File

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

12
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/Device.java
View File

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

26
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceAllShortOutput.java
View File

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

26
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceShortOutput.java
View File

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

2
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java
View File

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

16
fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceMapper.xml
View File

@ -36,6 +36,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="transport" column="transport" />
<result property="guid" column="guid" />
<result property="protocolCode" column="protocol_code" />
<result property="imei" column="imei" />
<result property="iccid" column="iccid" />
</resultMap>
<resultMap type="com.fastbee.iot.model.DeviceShortOutput" id="DeviceShortResult">
@ -64,6 +66,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="protocolCode" column="protocol_code"/>
<result property="transport" column="transport" />
<result property="guid" column="guid" />
<result property="imei" column="imei" />
<result property="iccid" column="iccid" />
</resultMap>
<resultMap type="com.fastbee.iot.model.DeviceAllShortOutput" id="DeviceAllShortResult">
@ -84,6 +88,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="latitude" column="latitude" />
<result property="isOwner" column="is_owner" />
<result property="subDeviceCount" column="sub_device_count"/>
<result property="imei" column="imei" />
<result property="iccid" column="iccid" />
</resultMap>
<resultMap type="com.fastbee.iot.model.UserAndTenant" id="UserAndTenantResult">
@ -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}
</select>
@ -657,6 +667,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="remark != null">remark = #{remark},</if>
<if test="imgUrl != null">img_url = #{imgUrl},</if>
<if test="summary != null">summary = #{summary},</if>
<if test="imei != null">imei = #{imei},</if>
<if test="iccid != null">iccid = #{iccid},</if>
<if test="slaveId != null">slave_id = #{slaveId},</if>
</trim>
where device_id = #{deviceId}
@ -718,6 +730,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="imgUrl != null">img_url = #{imgUrl},</if>
<if test="summary != null">summary = #{summary},</if>
<if test="gwDevCode != null">gw_dev_code = #{gwDevCode},</if>
<if test="imei != null">imei = #{imei},</if>
<if test="iccid != null">iccid = #{iccid},</if>
</trim>
where serial_number = #{serialNumber}
</update>

142
yack-web/src/views/iot/device/running-status.vue
View File

@ -21,6 +21,39 @@
<el-button type="success" size="mini" style="float: right" @click="getLatestFirmware(deviceInfo.deviceId)" :disabled="deviceInfo.status != 3">{{ $t('device.running-status.866086-2') }}</el-button>
</el-descriptions-item>
<!-- IMEI / ICCID（蜂窝信息） -->
<el-descriptions-item :labelStyle="statusColor">
<template slot="label">
<i class="el-icon-postcard"></i>
IMEI
</template>
<div class="cell-flex">
<el-input v-model="deviceInfo.imei" size="mini" placeholder="请输入IMEI" style="max-width: 260px;" @keyup.enter.native="saveImeiIccid('imei')" />
<el-popover placement="bottom" width="360" trigger="click" popper-class="atcmd-popover">
<div class="atcmd-title">编辑下发指令</div>
<el-input v-model="imeiCmd" size="mini" placeholder="例如：AT+IMEI\\r\\n" />
<el-button slot="reference" size="mini" icon="el-icon-edit" title="编辑指令"></el-button>
</el-popover>
<el-button type="primary" size="mini" :disabled="deviceInfo.status != 3" @click="sendAt('imei')" :loading="imeiLoading" title="默认指令:AT+IMEI\r\n">从设备读取</el-button>
</div>
</el-descriptions-item>
<el-descriptions-item :labelStyle="statusColor">
<template slot="label">
<i class="el-icon-postcard"></i>
ICCID
</template>
<div class="cell-flex">
<el-input v-model="deviceInfo.iccid" size="mini" placeholder="请输入ICCID" style="max-width: 260px;" @keyup.enter.native="saveImeiIccid('iccid')"/>
<el-popover placement="bottom" width="360" trigger="click" popper-class="atcmd-popover">
<div class="atcmd-title">编辑下发指令</div>
<el-input v-model="iccidCmd" size="mini" placeholder="例如：AT+ICCID\\r\\n" />
<el-button slot="reference" size="mini" icon="el-icon-edit" title="编辑指令"></el-button>
</el-popover>
<el-button type="primary" size="mini" :disabled="deviceInfo.status != 3" @click="sendAt('iccid')" :loading="iccidLoading" title="默认指令:AT+ICCID\r\n">从设备读取</el-button>
</div>
</el-descriptions-item>
<!-- 设备物模型-->
<el-descriptions-item v-for="(item, index) in deviceInfo.thingsModels" :key="index" :labelStyle="statusColor">
<template slot="label">
@ -512,6 +545,8 @@
import { getLatestFirmware } from '@/api/iot/firmware';
import { serviceInvoke } from '@/api/iot/runstatus';
import { messagePost } from '@/api/iot/mqttTest';
import { getDevice, updateDevice } from '@/api/iot/device';
export default {
name: 'running-status',
@ -581,6 +616,12 @@ export default {
],
resizeObserver: null,
remoteCommand: {},
imeiCmd: 'AT+IMEI\\r\\n',
iccidCmd: 'AT+ICCID\\r\\n',
imeiLoading: false,
iccidLoading: false,
saveLoadingImei: false,
saveLoadingIccid: false,
};
},
mounted() {
@ -600,6 +641,73 @@ export default {
},
methods: {
async refreshDeviceBaseInfo() {
if (!this.deviceInfo || !this.deviceInfo.deviceId) return;
const res = await getDevice(this.deviceInfo.deviceId);
if (res && res.code === 200 && res.data) {
// 只刷新这两个字段，避免影响运行态数据
this.deviceInfo.imei = res.data.imei;
this.deviceInfo.iccid = res.data.iccid;
}
},
async sendAt(type) {
try {
const cmd = type === 'imei' ? this.imeiCmd : this.iccidCmd;
if (!cmd) {
this.$message({ type: 'warning', message: '指令不能为空' });
return;
}
if (type === 'imei') this.imeiLoading = true;
if (type === 'iccid') this.iccidLoading = true;
await messagePost({
message: cmd,
serialNumber: this.deviceInfo.serialNumber,
topicName: '/function/get',
dataType: 'ascii',
});
this.$message({ type: 'success', message: '已下发指令，等待设备回复解析…' });
// 后端收到回复后会自动解析并写入数据库，这里做一次轻量轮询刷新
setTimeout(() => this.refreshDeviceBaseInfo(), 1200);
setTimeout(() => this.refreshDeviceBaseInfo(), 2500);
} catch (e) {
this.$message({ type: 'error', message: e && e.message ? e.message : '指令下发失败' });
} finally {
this.imeiLoading = false;
this.iccidLoading = false;
}
},
async saveImeiIccid(type) {
try {
if (!this.deviceInfo || !this.deviceInfo.deviceId) return;
if (type === 'imei') this.saveLoadingImei = true;
if (type === 'iccid') this.saveLoadingIccid = true;
const payload = {
deviceId: this.deviceInfo.deviceId,
serialNumber: this.deviceInfo.serialNumber,
};
if (type === 'imei') payload.imei = this.deviceInfo.imei;
if (type === 'iccid') payload.iccid = this.deviceInfo.iccid;
const res = await updateDevice(payload);
if (res && res.code === 200) {
this.$message({ type: 'success', message: '保存成功' });
await this.refreshDeviceBaseInfo();
} else {
this.$message({ type: 'error', message: (res && res.msg) || '保存失败' });
}
} catch (e) {
this.$message({ type: 'error', message: e && e.message ? e.message : '保存失败' });
} finally {
this.saveLoadingImei = false;
this.saveLoadingIccid = false;
}
},
initResizeObserver() {
if (this.resizeObserver) return;
if (typeof window.ResizeObserver !== 'function') return;
@ -735,10 +843,10 @@ export default {
if (data[j].id == this.monitorChart[m].data.id) {
// 更新数据对象
this.monitorChart[m].data.value = data[j].value;
// 确保数值是数字类型
const numValue = Number(data[j].value) || 0;
// 构建图表数据
const chartData = [
{
@ -746,7 +854,7 @@ export default {
name: this.monitorChart[m].data.name,
},
];
// 更新图表配置，包括detail部分
this.monitorChart[m].chart.setOption({
series: [
@ -758,14 +866,14 @@ export default {
},
],
}, false); // false表示不合并，完全替换
// 强制刷新图表
this.$nextTick(() => {
if (this.monitorChart[m] && this.monitorChart[m].chart) {
this.monitorChart[m].chart.resize();
}
});
break;
}
}
@ -779,10 +887,10 @@ export default {
if (data[j].id == this.monitorChart[m].data.id) {
// 更新数据对象
this.monitorChart[m].data.value = data[j].value;
// 确保数值是数字类型
const numValue = Number(data[j].value) || 0;
// 构建图表数据
const chartData = [
{
@ -790,7 +898,7 @@ export default {
name: this.monitorChart[m].data.name,
},
];
// 更新图表配置，包括detail部分
this.monitorChart[m].chart.setOption({
series: [
@ -802,14 +910,14 @@ export default {
},
],
}, false); // false表示不合并，完全替换
// 强制刷新图表
this.$nextTick(() => {
if (this.monitorChart[m] && this.monitorChart[m].chart) {
this.monitorChart[m].chart.resize();
}
});
break;
}
}
@ -1083,6 +1191,20 @@ export default {
padding-left: 20px;
}
.cell-flex {
display: flex;
align-items: center;
gap: 10px;
flex-wrap: wrap;
}
.atcmd-title {
font-size: 13px;
font-weight: 600;
color: #303133;
margin-bottom: 10px;
}
.monitor-grid-wrap {
background-color: #f5f7fa;
padding: 20px 10px;