package org.jeecg.modules.quartz.job;
|
|
import com.influxdb.client.InfluxDBClient;
|
import com.influxdb.client.InfluxDBClientFactory;
|
import com.influxdb.client.QueryApi;
|
import com.influxdb.query.FluxTable;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.collections.CollectionUtils;
|
import org.jeecg.modules.iot.depository.Influxdb;
|
import org.jeecg.modules.iot.depository.InfluxdbTest;
|
import org.jeecg.modules.iot.entity.Equipment;
|
import org.jeecg.modules.iot.entity.InfluxdbDeploy;
|
import org.jeecg.modules.iot.entity.ServerDeploy;
|
import org.jeecg.modules.iot.mdc.entity.EquipmentAlarm;
|
import org.jeecg.modules.iot.mdc.entity.EquipmentLog;
|
import org.jeecg.modules.iot.mdc.service.IEquipmentAlarmService;
|
import org.jeecg.modules.iot.mdc.service.IEquipmentLogService;
|
import org.jeecg.modules.iot.service.IEquipmentService;
|
import org.jeecg.modules.iot.service.IInfluxdbDeployService;
|
import org.quartz.Job;
|
import org.quartz.JobExecutionContext;
|
import org.quartz.JobExecutionException;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Value;
|
|
import javax.sql.DataSource;
|
import java.sql.Connection;
|
import java.sql.SQLException;
|
import java.text.SimpleDateFormat;
|
import java.time.LocalDateTime;
|
import java.time.ZoneId;
|
import java.time.format.DateTimeFormatter;
|
import java.time.temporal.TemporalAccessor;
|
import java.util.*;
|
import java.util.concurrent.atomic.AtomicReference;
|
import java.util.stream.Collectors;
|
|
/**
|
* influxdb定时任务
|
*
|
* @Author Scott
|
*/
|
@Slf4j
|
public class SampleParamJob implements Job {
|
@Value("${databaseType}")
|
private String databaseType;
|
@Autowired
|
private IInfluxdbDeployService influxdbDeployService;
|
@Autowired
|
private DataSource dataSource;
|
@Autowired
|
private IEquipmentService equipmentService;
|
@Autowired
|
private IEquipmentLogService equipmentLogService;
|
@Autowired
|
private org.jeecg.modules.iot.mdc.service.IEquipmentService mdcEquipmentService;
|
@Autowired
|
private IEquipmentAlarmService equipmentAlarmService;
|
@Autowired
|
private org.jeecg.modules.iot.service.IServerDeployService serverDeployService;
|
|
@Override
|
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
|
List<InfluxdbDeploy> influxdbDeployList = influxdbDeployService.list();
|
List<ServerDeploy> serverDeployList = serverDeployService.list();
|
|
influxdbDeployList.forEach(in -> {
|
// InfluxDB 2.0 连接信息
|
String influxUrl = "http://" + in.getAddress() + ":" + in.getPort();
|
String influxToken = in.getAuthorizeCode();
|
String influxOrg = in.getOrganization();
|
String influxBucket = in.getBucket();
|
// 连接 InfluxDB 2.0
|
InfluxDBClient influxDBClient = InfluxDBClientFactory.create(influxUrl, influxToken.toCharArray());
|
QueryApi queryApi = influxDBClient.getQueryApi();
|
// 连接 SqlServer
|
try (Connection mysqlConnection = dataSource.getConnection()) {
|
// 处理时间
|
DateTimeFormatter inputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
AtomicReference<String> start = new AtomicReference<>("");
|
AtomicReference<String> end = new AtomicReference<>("");
|
AtomicReference<Date> endDate = new AtomicReference<>(null); // 采集时间
|
serverDeployList.forEach(s -> {
|
if (s.getId().equals(in.getServerDeployId()) && s.getCollectTime() != null) {
|
start.set(sdf.format(s.getCollectTime()));
|
// 获取当前时间
|
LocalDateTime now = LocalDateTime.now();
|
// 减去一分钟
|
// LocalDateTime oneMinuteAgo = now.minusMinutes(1);
|
LocalDateTime oneMinuteAgo = now.minusSeconds(2);
|
// 格式化为字符串（可选）
|
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
String formattedTime = oneMinuteAgo.format(formatter);
|
// 转换为 Date 类型（使用系统默认时区）
|
Date date = Date.from(oneMinuteAgo.atZone(ZoneId.systemDefault()).toInstant());
|
endDate.set(date);
|
end.set(formattedTime);
|
}
|
});
|
if (start.get().equals("")){
|
return;
|
}
|
LocalDateTime startTime = LocalDateTime.parse(start.get(), inputFormatter);
|
LocalDateTime endTime = LocalDateTime.parse(end.get(), inputFormatter);
|
// 定义日期时间格式器（ISO_LOCAL_DATE_TIME 对应 "yyyy-MM-ddTHH:mm:ss"）
|
DateTimeFormatter formatter1 = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
|
|
// 解析字符串为 LocalDateTime 对象
|
LocalDateTime startDateTime = LocalDateTime.parse(startTime.toString(), formatter1);// 解析字符串为 LocalDateTime 对象
|
LocalDateTime endDateTime = LocalDateTime.parse(endTime.toString(), formatter1);
|
|
// 加上 8 小时
|
LocalDateTime startNewDateTime = startDateTime.minusHours(8);
|
LocalDateTime endNewDateTime = endDateTime.minusHours(8);
|
// String query = "from(bucket: \"IOT\") \n" +
|
// "|> range(start: "+startNewDateTime+"Z, stop: "+endNewDateTime+"Z) \n" +
|
// "|> fill(usePrevious: true) \n" +
|
// "|> filter(fn: (r) => r[\"_field\"] == \"TimeStamp\" or r[\"_field\"] == \"Value\")";
|
String query = "from(bucket: \"" + influxBucket + "\") " +
|
"|> range(start: -5s) " +
|
"|> filter(fn: (r) => r[\"_field\"] == \"TimeStamp\" or r[\"_field\"] == \"Value\")";
|
|
// 执行查询
|
try {
|
List<FluxTable> tables = queryApi.query(query, influxOrg);
|
// 处理查询结果并插入到 MySQL
|
List<Influxdb> influxdbs = InfluxdbTest.processAndInsert(tables, mysqlConnection);
|
// 根据类型拆分
|
Map<String, List<Influxdb>> fieldistMap = influxdbs.stream()
|
.collect(Collectors.groupingBy(Influxdb::getField));
|
List<Influxdb> valueList = new ArrayList<>();
|
List<Influxdb> timeStampList = new ArrayList<>();
|
fieldistMap.forEach((field, timeList) -> {
|
if (field.equals("Value")) {
|
valueList.addAll(timeList);
|
} else {
|
timeStampList.addAll(timeList);
|
}
|
});
|
for (int i = 0; i < valueList.size(); i++) {
|
valueList.get(i).setAcquisitionTime(timeStampList.get(i).getValue());
|
}
|
// 根据设备进行分组
|
Map<String, List<Influxdb>> measurementListMap = valueList.stream()
|
.collect(Collectors.groupingBy(Influxdb::getMeasurement));
|
measurementListMap.forEach((table, timeList) -> {
|
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
Map<String, List<Influxdb>> timeListMap = timeList.stream()
|
.collect(Collectors.groupingBy(
|
Influxdb::getAcquisitionTime,
|
() -> new TreeMap<>(Comparator.comparing(
|
timeStr -> LocalDateTime.parse(timeStr, formatter)
|
)),
|
Collectors.toList()
|
));
|
|
// 处理设备状态
|
int lastIndex = table.lastIndexOf('.');
|
String code = table.substring(lastIndex + 1);
|
EquipmentLog equipmentLog = equipmentLogService.selectEquipmentOporation(code);
|
// 设备状态
|
Integer equipmentState = null;
|
for (Map.Entry<String, List<Influxdb>> entry : timeListMap.entrySet()) {
|
String measurement = entry.getKey();
|
List<Influxdb> influxdbList = entry.getValue();
|
Equipment equipment = equipmentService.findEquipmentByCode(code);
|
String tableName = equipment.getControlSystem() + '_' + equipment.getEqptCode();
|
// 获取表名
|
String[] columns = new String[influxdbList.size() + 2];
|
Object[] values = new Object[influxdbList.size() + 2];
|
String state = ""; // 设备状态
|
String isAlarm = ""; // 报警状态
|
String alarm = ""; // 报警号
|
String alarmContent = ""; // 报警内容
|
// 遍历当前设备下的所有参数
|
for (int i = 0; i < influxdbList.size(); i++) {
|
String parameter = influxdbList.get(i).getParameter();
|
columns[i] = parameter.substring(parameter.lastIndexOf('-') + 1);
|
values[i] = influxdbList.get(i).getValue();
|
if (values[i] == null || values[i] == "") {
|
values[i] = "";
|
} else {
|
if (columns[i].equals("Oporation")) {
|
state = values[i].toString();
|
} else if (columns[i].equals("AlarmNo")) {
|
alarm = values[i].toString();
|
} else if (columns[i].equals("AlarmContent")) {
|
alarmContent = values[i].toString();
|
} else if (columns[i].equals("IsAlarm")) {
|
isAlarm = values[i].toString();
|
}
|
}
|
|
}
|
// 验证数据是否都为空
|
if (!allEmptyStrings(values)) {
|
columns[influxdbList.size() + 1] = "CollectTime";
|
columns[influxdbList.size()] = "EquipmentID";
|
// 转换为 LocalDateTime
|
LocalDateTime dateTime = LocalDateTime.parse(influxdbList.get(0).getAcquisitionTime(), formatter);
|
values[influxdbList.size() + 1] = dateTime;
|
values[influxdbList.size()] = equipment.getEqptCode();
|
// 插入数据
|
if (databaseType.equals("SqlServer")) {
|
equipmentService.insertSqlServerData(tableName, columns, values);
|
} else if (databaseType.equals("MySql")) {
|
equipmentService.insertMySqlData(tableName, columns, values);
|
|
}
|
if (!state.equals("")) {
|
EquipmentLog log = new EquipmentLog();
|
log.setOporation(Integer.parseInt(state));
|
log.setEquipmentId(equipment.getEqptCode());
|
log.setEquipmentName(equipment.getEqptName());
|
log.setCollectTime(Date.from(dateTime.atZone(ZoneId.systemDefault()).toInstant()));
|
// True 报警开始，False报警结束
|
if (isAlarm.equals("True")) {
|
// 报警开始
|
if (equipmentLog == null && equipmentState == null) {
|
// 验证是否报警
|
log.setOporation(22);
|
log.setAlarm(alarm);
|
equipmentLogService.save(log);
|
equipmentState = 22;
|
equipmentStateModification(equipment.getEqptCode(), 22);
|
} else if (equipmentLog != null && equipmentState == null && equipmentLog.getOporation() != 22) {
|
log.setOporation(22);
|
log.setAlarm(alarm);
|
equipmentState = 22;
|
equipmentLogService.save(log);
|
equipmentStateModification(equipment.getEqptCode(), 22);
|
} else if (equipmentLog != null && equipmentState != null && equipmentState != 22) {
|
log.setOporation(22);
|
log.setAlarm(alarm);
|
equipmentState = 22;
|
equipmentLogService.save(log);
|
equipmentStateModification(equipment.getEqptCode(), 22);
|
}
|
} else if (isAlarm.equals("False")) {
|
// 报警结束
|
if (equipmentLog == null && equipmentState == null) {
|
// 验证是否报警
|
equipmentLogService.save(log);
|
equipmentState = Integer.parseInt(state);
|
equipmentStateModification(equipment.getEqptCode(), equipmentState);
|
} else if (equipmentLog != null && equipmentState == null && equipmentLog.getOporation() == 22) {
|
log.setOporation(23);
|
equipmentState = 23;
|
equipmentLogService.save(log);
|
equipmentStateModification(equipment.getEqptCode(), equipmentState);
|
} else if (equipmentLog != null && equipmentState != null && equipmentState == 22) {
|
log.setOporation(23);
|
equipmentState = 23;
|
equipmentLogService.save(log);
|
equipmentStateModification(equipment.getEqptCode(), 23);
|
} else {
|
if (equipmentLog != null && equipmentState == null && equipmentLog.getOporation() != Integer.parseInt(state)) {
|
equipmentState = Integer.parseInt(state);
|
equipmentLogService.save(log);
|
equipmentStateModification(equipment.getEqptCode(), equipmentState);
|
//验证是否是报警，或者报警结束
|
} else if (equipmentLog != null && equipmentState != null && equipmentState != Integer.parseInt(state)) {
|
equipmentState = Integer.parseInt(state);
|
equipmentLogService.save(log);
|
equipmentStateModification(equipment.getEqptCode(), equipmentState);
|
}
|
}
|
}
|
// 处理报警信息
|
if (isAlarm.equals("True")) {
|
EquipmentAlarm equipmentAlarm = equipmentAlarmService.selectEquipmentAlarmByEquipmentId(equipment.getEqptCode());
|
if (equipmentAlarm != null && !equipmentAlarm.getAlarmNo().equals(alarm)) {
|
addEquipmentAlarm(alarm, equipment, dateTime, alarmContent);
|
} else if (equipmentAlarm == null) {
|
addEquipmentAlarm(alarm, equipment, dateTime, alarmContent);
|
}
|
}
|
}
|
}
|
}
|
});
|
// 如果有查询数据，时间不为空
|
|
if (endDate.get() != null && CollectionUtils.isNotEmpty(tables)) {
|
for (ServerDeploy s : serverDeployList) {
|
if (s.getId().equals(in.getServerDeployId())) {
|
s.setCollectTime(endDate.get());
|
return;
|
}
|
}
|
}
|
} catch (Exception e) {
|
log.error(String.valueOf(e));
|
}
|
} catch (SQLException e) {
|
e.printStackTrace();
|
} finally {
|
influxDBClient.close();
|
}
|
});
|
serverDeployService.updateBatchById(serverDeployList);
|
// 直接在输出语句中格式化
|
log.info("实时数据任务执行，当前时间：" +
|
LocalDateTime.ofInstant(new Date().toInstant(), ZoneId.systemDefault())
|
.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")));
|
}
|
|
public static boolean allEmptyStrings(Object[] values) {
|
if (values == null) {
|
return false; // 如果数组本身为null，返回false
|
}
|
for (Object str : values) {
|
if (str != null && !str.toString().equals("")) {
|
return false; // 如果元素不是String类型，返回false
|
}
|
}
|
|
return true; // 所有元素都是空字符串
|
}
|
|
/**
|
* 报警信息处理
|
*
|
* @param alarm 报警号
|
* @param equipment 设备信息
|
* @param dateTime 报警时间
|
* @param alarmContent 报警内容
|
*/
|
public void addEquipmentAlarm(String alarm, Equipment equipment, LocalDateTime dateTime, String alarmContent) {
|
// 验证是否报警
|
EquipmentAlarm equipmentAlarm = new EquipmentAlarm();
|
equipmentAlarm.setEquipmentid(equipment.getEqptCode());
|
equipmentAlarm.setCollecttime(Date.from(dateTime.atZone(ZoneId.systemDefault()).toInstant()));
|
equipmentAlarm.setAlarmNo(alarm);
|
equipmentAlarm.setAlarmContent(alarmContent);
|
equipmentAlarmService.save(equipmentAlarm);
|
|
}
|
|
void equipmentStateModification(String eqptCode, Integer state) {
|
org.jeecg.modules.iot.mdc.entity.Equipment mdcEquipment = mdcEquipmentService.findEquipmentByEquipmentId(eqptCode);
|
if (mdcEquipment != null) {
|
mdcEquipment.setOporation(state);
|
mdcEquipmentService.updateById(mdcEquipment);
|
}
|
}
|
}
|
