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

20260915:traceId日志链路追踪

Browse Source
This commit is contained in:
zhugongping 2026-09-16 15:07:16 +08:00
parent d7a524cba3
commit f30ffda889
15 changed files with 64 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
fastbee-common/src/main/java/com/fastbee/common/utils/StringUtils.java
View File

@ -295,21 +295,19 @@ public class StringUtils extends org.apache.commons.lang3.StringUtils {
/**
* 判断给定的set列表中是否包含数组array 判断给定的数组array中是否包含给定的元素value
*
* @param set 给定的集合
* @param collection 给定的集合
* @param array 给定的数组
* @return boolean 结果
*/
public static boolean containsAny(Collection<String> collection, String... array) {
if (isEmpty(collection) || isEmpty(array)) {
return false;
} else {
if (!isEmpty(collection) && !isEmpty(array)) {
for (String str : array) {
if (collection.contains(str)) {
return true;
}
}
return false;
}
return false;
}
/**

2
fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java
View File

@ -99,7 +99,7 @@ public class DevicePropFetchConsumer {
// 不进行格式校验，直接转换数据
byte[] commandData = ByteUtils.convertToHexByteArray(command);
switch (serverType) {
//通过mqtt内部客户端 下发指令
// 通过mqtt内部客户端 下发指令
case MQTT:
publish(topic, commandData);
log.info("=>MQTT-线程=[{}],设备=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), serialNumber, command, topic);

13
fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceAbstractQuartzJob.java
View File

@ -30,7 +30,7 @@ public abstract class DeviceAbstractQuartzJob implements Job {
/**
* 线程本地变量
*/
private static ThreadLocal<Date> threadLocal = new ThreadLocal<>();
private static final ThreadLocal<Date> threadLocal = new ThreadLocal<>();
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
@ -38,9 +38,7 @@ public abstract class DeviceAbstractQuartzJob implements Job {
BeanUtils.copyBeanProp(deviceJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES));
try {
before(context, deviceJob);
if (deviceJob != null) {
doExecute(context, deviceJob);
}
after(context, deviceJob, null);
} catch (Exception e) {
log.error("任务执行异常 - ：", e);
@ -92,10 +90,6 @@ public abstract class DeviceAbstractQuartzJob implements Job {
startTime = new Date();
}
sysJobLog.setStartTime(startTime);
Date stopTime = new Date();
sysJobLog.setStopTime(stopTime);
long runMs = stopTime.getTime() - startTime.getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒");
if (e != null) {
sysJobLog.setStatus(Constants.FAIL);
String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000);
@ -104,6 +98,11 @@ public abstract class DeviceAbstractQuartzJob implements Job {
sysJobLog.setStatus(Constants.SUCCESS);
}
Date stopTime = new Date();
sysJobLog.setStopTime(stopTime);
long runMs = stopTime.getTime() - startTime.getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒");
// 写入数据库当中
SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog);

1
fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java
View File

@ -11,6 +11,7 @@ import org.quartz.JobExecutionContext;
*/
@DisallowConcurrentExecution
public class DeviceDisallowConcurrentExecution extends DeviceAbstractQuartzJob {
@Override
protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception {
DeviceJobInvoke.invokeMethod(deviceJob);

1
fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java
View File

@ -9,6 +9,7 @@ import org.quartz.JobExecutionContext;
* @author sumuxi, 2026-04-16
*/
public class DeviceJobExecution extends DeviceAbstractQuartzJob {
@Override
protected void doExecute(JobExecutionContext context, DeviceJob deviceJob) throws Exception {
DeviceJobInvoke.invokeMethod(deviceJob);

38
fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobInvoke.java
View File

@ -29,13 +29,13 @@ public class DeviceJobInvoke {
/**
* 获取消息推送接口
*/
private static IMqttMessagePublish messagePublish = SpringUtils.getBean(IMqttMessagePublish.class);
private static final IMqttMessagePublish messagePublish = SpringUtils.getBean(IMqttMessagePublish.class);
private static SceneMapper sceneMapper = SpringUtils.getBean(SceneMapper.class);
private static final SceneMapper sceneMapper = SpringUtils.getBean(SceneMapper.class);
private static FlowLogExecutor flowExecutor = SpringUtils.getBean(FlowLogExecutor.class);
private static final FlowLogExecutor flowExecutor = SpringUtils.getBean(FlowLogExecutor.class);
private static IDataHandler dataHandler = SpringUtils.getBean(IDataHandler.class);
private static final IDataHandler dataHandler = SpringUtils.getBean(IDataHandler.class);
/**
* 执行方法
@ -44,45 +44,45 @@ public class DeviceJobInvoke {
*/
public static void invokeMethod(DeviceJob deviceJob) throws Exception {
if (deviceJob.getJobType() == 1) {
log.info("------------------------1、执行设备定时任务-----------------------------");
log.info("1、[执行设备定时任务]");
List<Action> actions = JSON.parseArray(deviceJob.getActions(), Action.class);
List<ThingsModelSimpleItem> propertys = new ArrayList<>();
List<ThingsModelSimpleItem> properties = new ArrayList<>();
List<ThingsModelSimpleItem> functions = new ArrayList<>();
for (int i = 0; i < actions.size(); i++) {
for (Action action : actions) {
ThingsModelSimpleItem model = new ThingsModelSimpleItem();
model.setId(actions.get(i).getId());
model.setValue(actions.get(i).getValue());
model.setId(action.getId());
model.setValue(action.getValue());
model.setRemark("设备定时");
if (actions.get(i).getType() == 1) {
propertys.add(model);
} else if (actions.get(i).getType() == 2) {
if (action.getType() == 1) {
properties.add(model);
} else if (action.getType() == 2) {
functions.add(model);
}
}
// 发布属性
if (propertys.size() > 0) {
messagePublish.publishProperty(deviceJob.getProductId(), deviceJob.getSerialNumber(), propertys, 0);
if (!properties.isEmpty()) {
messagePublish.publishProperty(deviceJob.getProductId(), deviceJob.getSerialNumber(), properties, 0);
}
// 发布功能
if (functions.size() > 0) {
if (!functions.isEmpty()) {
messagePublish.publishFunction(deviceJob.getProductId(), deviceJob.getSerialNumber(), functions, 0);
}
} else if (deviceJob.getJobType() == 3) {
log.info("------------------3、[定时执行场景联动]---------------------");
log.info("3、[定时执行场景联动]");
Scene scene = sceneMapper.selectSceneBySceneId(deviceJob.getSceneId());
// 执行场景规则,异步非阻塞
SceneContext context = new SceneContext("", 0L, 0, null);
String requestId = "scene/" + scene.getChainName();
flowExecutor.execute2FutureWithRid(scene.getChainName(), null, requestId, context);
} else if (4 == deviceJob.getJobType()) {
log.info("------------------4、[定时执行场景运算型变量]---------------------");
log.info("4、[定时执行场景运算型变量]");
String s = dataHandler.calculateSceneModelTagValue(deviceJob.getDatasourceId());
if (StringUtils.isEmpty(s)) {
log.info("------------------[定时执行场景运算型变量失败：+{}]---------------------", s);
log.info("[定时执行场景运算型变量失败：+{}]", s);
}
} else if (5 == deviceJob.getJobType()) {
log.info("----------------5、执行modbus轮训指令----------------------");
log.info("5、[执行Modbus轮训指令]");
ModbusPollJob job = new ModbusPollJob();
job.setType(2);
job.setDeviceJob(deviceJob);

2
fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/SyncDeviceStatusJob.java
View File

@ -38,7 +38,7 @@ public class SyncDeviceStatusJob {
*/
public void syncDeviceStatus() {
if (enabled) {
//获取所有已激活并不是禁用的设备
// 获取所有已激活并不是禁用的设备
List<DeviceStatusVO> deviceStatusVOList = deviceService.selectDeviceActive();
if (!CollectionUtils.isEmpty(deviceStatusVOList)) {
for (DeviceStatusVO statusVO : deviceStatusVOList) {

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

@ -40,6 +40,9 @@ public class DeviceReplyListen {
}
} catch (Exception e) {
log.error("=>设备回调消息监听异常", e);
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
}
}
}

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

@ -37,6 +37,9 @@ public class DeviceTestListen {
}
} catch (Exception e) {
log.error("=>emq数据转发异常");
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
}
}
}

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

@ -38,6 +38,9 @@ public class FunctionInvokeListen {
}
} catch (Exception e) {
log.error("=>下发服务消费异常", e);
} finally {
// 确保任何情况下都清理上下文
TraceContext.clear();
}
}
}

4
fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java
View File

@ -47,6 +47,10 @@ public class DeviceJobServiceImpl implements IDeviceJobService {
@PostConstruct
public void init() throws SchedulerException, TaskException {
scheduler.clear();
// 打印实际类型
// log.info("Scheduler实际类型: {}", scheduler.getClass().getName());
// log.info("Scheduler元数据: {}", scheduler.getMetaData().getSummary());
// log.info("Scheduler class: {}", scheduler.getClass());
// 设备定时任务
List<DeviceJob> jobList = jobMapper.selectJobAll();
for (DeviceJob deviceJob : jobList) {

8
fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java
View File

@ -222,7 +222,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
/**
* 变量读取
*
* @param readVO
* @param readVO 前端入参
*/
@Override
public void readVariableValue(VariableReadVO readVO) {
@ -233,7 +233,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
deviceAndProtocol.getProtocolCode().equals(FastBeeConstant.PROTOCOL.ModbusTcp)))
throw new ServiceException("非modbus协议请先配置主动采集协议");
Long productId = deviceAndProtocol.getProductId();
//如果是子设备，则获取网关子设备的产品id和设备编号
// 如果是子设备，则获取网关子设备的产品id和设备编号
Integer deviceType = deviceAndProtocol.getDeviceType();
if (deviceType == DeviceType.SUB_GATEWAY.getCode()) {
serialNumber = deviceAndProtocol.getGwSerialNumber();
@ -242,7 +242,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
Integer type = readVO.getType();
type = Objects.isNull(type) ? 1 : type;
if (type == 1) {
//单个变量获取
// 单个变量获取
String identifier = readVO.getIdentifier();
ThingsModelValueItem thingModels = itslCache.getSingleThingModels(deviceAndProtocol.getProductId(), identifier);
ModbusConfig config = thingModels.getConfig();
@ -275,7 +275,7 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
job.setType(1);
MessageProducer.sendPropFetch(job);
} else {
//读取当前设备的所有变量,这里读取所有的，判断传递的设备是网关设备、网关子设备、直连设备
// 读取当前设备的所有变量,这里读取所有的，判断传递的设备是网关设备、网关子设备、直连设备
DeviceType code = DeviceType.transfer(deviceType);
List<ModbusJob> modbusJobList = modbusJobService.selectDevicesJobByDeviceType(code, readVO.getSerialNumber());
List<String> commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList());

8
fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/controller/SysJobController.java
View File

@ -83,9 +83,9 @@ public class SysJobController extends BaseController {
return error(StringUtils.format(MessageUtils.message("job.add.failed.cron.not.valid"), job.getJobName()));
} else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.rmi.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS})) {
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.ldap.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.HTTP, Constants.HTTPS})) {
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.HTTP, Constants.HTTPS)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.http.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) {
return error(StringUtils.format(MessageUtils.message("job.add.failed.string.error"), job.getJobName()));
@ -108,9 +108,9 @@ public class SysJobController extends BaseController {
return error(StringUtils.format(MessageUtils.message("job.update.failed.cron.not.valid"), job.getJobName()));
} else if (StringUtils.containsIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_RMI)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.rmi.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS})) {
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.LOOKUP_LDAP, Constants.LOOKUP_LDAPS)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.ldap.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), new String[]{Constants.HTTP, Constants.HTTPS})) {
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.HTTP, Constants.HTTPS)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.http.not.valid"), job.getJobName()));
} else if (StringUtils.containsAnyIgnoreCase(job.getInvokeTarget(), Constants.JOB_ERROR_STR)) {
return error(StringUtils.format(MessageUtils.message("job.update.failed.string.error"), job.getJobName()));

14
fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/AbstractQuartzJob.java
View File

@ -30,7 +30,7 @@ public abstract class AbstractQuartzJob implements Job {
/**
* 线程本地变量
*/
private static ThreadLocal<Date> threadLocal = new ThreadLocal<>();
private static final ThreadLocal<Date> threadLocal = new ThreadLocal<>();
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
@ -38,9 +38,7 @@ public abstract class AbstractQuartzJob implements Job {
BeanUtils.copyBeanProp(sysJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES));
try {
before(context, sysJob);
if (sysJob != null) {
doExecute(context, sysJob);
}
after(context, sysJob, null);
} catch (Exception e) {
log.error("任务执行异常 - ：", e);
@ -78,9 +76,6 @@ public abstract class AbstractQuartzJob implements Job {
sysJobLog.setJobGroup(sysJob.getJobGroup());
sysJobLog.setInvokeTarget(sysJob.getInvokeTarget());
sysJobLog.setStartTime(startTime);
sysJobLog.setStopTime(new Date());
long runMs = sysJobLog.getStopTime().getTime() - sysJobLog.getStartTime().getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒");
if (e != null) {
sysJobLog.setStatus(Constants.FAIL);
String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000);
@ -88,14 +83,17 @@ public abstract class AbstractQuartzJob implements Job {
} else {
sysJobLog.setStatus(Constants.SUCCESS);
}
sysJobLog.setStopTime(new Date());
long runMs = sysJobLog.getStopTime().getTime() - sysJobLog.getStartTime().getTime();
sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒");
// 写入数据库当中
SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog);
log.info("定时任务执行完成: {} - {}, 耗时: {}ms", sysJob.getJobName(), sysJob.getJobGroup(), runMs);
// 清除 TraceContext
TraceContext.clear();
log.info("定时任务执行完成: {} - {}, 耗时: {}ms", sysJob.getJobName(), sysJob.getJobGroup(), runMs);
}
/**

6
fastbee-plugs/fastbee-quartz/src/main/java/com/fastbee/quartz/util/JobInvokeUtil.java
View File

@ -99,8 +99,8 @@ public class JobInvokeUtil {
}
String[] methodParams = methodStr.split(",(?=([^\"']*[\"'][^\"']*[\"'])*[^\"']*$)");
List<Object[]> classs = new LinkedList<>();
for (int i = 0; i < methodParams.length; i++) {
String str = StringUtils.trimToEmpty(methodParams[i]);
for (String methodParam : methodParams) {
String str = StringUtils.trimToEmpty(methodParam);
// String字符串类型，以'或"开头
if (StringUtils.startsWithAny(str, "'", "\"")) {
classs.add(new Object[]{StringUtils.substring(str, 1, str.length() - 1), String.class});
@ -151,7 +151,7 @@ public class JobInvokeUtil {
Object[] classs = new Object[methodParams.size()];
int index = 0;
for (Object[] os : methodParams) {
classs[index] = (Object) os[0];
classs[index] = os[0];
index++;
}
return classs;