xiaoguangbin/AnalysisSystemForRadionuclide
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

同步mdc代码

Browse Source
This commit is contained in:
qiaoqinzheng 2024-04-19 16:53:17 +08:00
parent 01122e6daa
commit b0529f0f3d
14 changed files with 189 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
jeecg-boot-base-core/src/main/java/org/jeecg/common/email/EmailServiceManager.java
View File

@ -669,7 +669,7 @@ public class EmailServiceManager {
String key = RedisConstant.EMAIL_MSG_ID+StringConstant.COLON+messageId; String key = RedisConstant.EMAIL_MSG_ID+StringConstant.COLON+messageId;
int numberKey = redisUtil.get(key) != null? (int) redisUtil.get(key):0; int numberKey = redisUtil.get(key) != null? (int) redisUtil.get(key):0;
// exist = redisUtil.hasKey(key); // exist = redisUtil.hasKey(key);
if(numberKey > taskProperties.getForceDeletedNumber()){ if(numberKey >= taskProperties.getForceDeletedNumber()){
exist = true; exist = true;
log.info("Check: Remove Email:{},receiveTime:{}",message.getSubject(), DateUtils.formatDate(message.getReceivedDate(),"yyyy-MM-dd HH:mm:ss")); log.info("Check: Remove Email:{},receiveTime:{}",message.getSubject(), DateUtils.formatDate(message.getReceivedDate(),"yyyy-MM-dd HH:mm:ss"));
message.setFlag(Flags.Flag.DELETED,true); message.setFlag(Flags.Flag.DELETED,true);

160
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/redisStream/AnalysisConsumer.java
View File

@ -5,8 +5,10 @@ import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil; import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.map.MapUtil; import cn.hutool.core.map.MapUtil;
import cn.hutool.core.text.StrBuilder;
import cn.hutool.core.util.NumberUtil; import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
@ -16,14 +18,15 @@ import org.jeecg.common.config.mqtoken.UserTokenContext;
import org.jeecg.common.constant.CommonConstant; import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.constant.SymbolConstant; import org.jeecg.common.constant.SymbolConstant;
import org.jeecg.common.constant.enums.SampleType; import org.jeecg.common.constant.enums.SampleType;
import org.jeecg.common.util.RedisStreamUtil; import org.jeecg.common.util.*;
import org.jeecg.common.util.SpringContextUtils; import org.jeecg.common.util.dynamic.db.FreemarkerParseFactory;
import org.jeecg.modules.base.dto.NuclideInfo; import org.jeecg.modules.base.dto.NuclideInfo;
import org.jeecg.modules.base.dto.Info; import org.jeecg.modules.base.dto.Info;
import org.jeecg.modules.base.entity.postgre.AlarmAnalysisLog; import org.jeecg.modules.base.entity.postgre.AlarmAnalysisLog;
import org.jeecg.modules.base.entity.postgre.AlarmAnalysisNuclideAvg; import org.jeecg.modules.base.entity.postgre.AlarmAnalysisNuclideAvg;
import org.jeecg.modules.base.entity.postgre.AlarmAnalysisRule; import org.jeecg.modules.base.entity.postgre.AlarmAnalysisRule;
import org.jeecg.modules.base.enums.Condition; import org.jeecg.modules.base.enums.Condition;
import org.jeecg.modules.base.enums.DSType;
import org.jeecg.modules.feignclient.SystemClient; import org.jeecg.modules.feignclient.SystemClient;
import org.jeecg.modules.service.AnalysisResultService; import org.jeecg.modules.service.AnalysisResultService;
import org.jeecg.modules.service.IAlarmAnalysisLogService; import org.jeecg.modules.service.IAlarmAnalysisLogService;
@ -36,9 +39,14 @@ import org.springframework.stereotype.Component;
import static org.jeecg.common.constant.enums.MessageTypeEnum.*; import static org.jeecg.common.constant.enums.MessageTypeEnum.*;
import static org.jeecg.common.util.TokenUtils.getTempToken; import static org.jeecg.common.util.TokenUtils.getTempToken;
import static org.jeecg.modules.base.enums.Template.ANALYSIS_NUCLIDE;
import static org.jeecg.modules.base.enums.Template.MONITOR_EMAIL;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.*; import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@Data @Data
@ -71,7 +79,7 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
try { try {
String streamKey = message.getStream(); String streamKey = message.getStream();
init(); init();
/** /*
* 新消息在未进行ACK之前,状态也为pending, * 新消息在未进行ACK之前,状态也为pending,
* 直接消费所有异常未确认的消息和新消息 * 直接消费所有异常未确认的消息和新消息
*/ */
@ -86,10 +94,10 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
// 消费完成后,手动确认消费消息[消息消费成功] // 消费完成后,手动确认消费消息[消息消费成功]
// 否则就是消费抛出异常,进入pending_ids[消息消费失败] // 否则就是消费抛出异常,进入pending_ids[消息消费失败]
redisStreamUtil.ack(streamKey, groupName, recordId.getValue()); redisStreamUtil.ack(streamKey, groupName, recordId.getValue());
// TODO del 取消手动删除已消费消息 // 手动删除已消费消息
// redisStreamUtil.del(streamKey, recordId.getValue()); redisStreamUtil.del(streamKey, recordId.getValue());
} }
}catch (RuntimeException e){ }catch (Exception e){
log.error("AnalysisConsumer消费异常: {}", e.getMessage()); log.error("AnalysisConsumer消费异常: {}", e.getMessage());
}finally { }finally {
destroy(); destroy();
@ -109,7 +117,8 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
for (AlarmAnalysisRule rule : rules) { for (AlarmAnalysisRule rule : rules) {
// 当前规则是否有报警条件 // 当前规则是否有报警条件
String conditionStr = rule.getConditions(); String conditionStr = rule.getConditions();
if (StrUtil.isBlank(conditionStr)) continue; if (StrUtil.isBlank(conditionStr))
continue;
// 是否在当前规则关注的台站列表内 // 是否在当前规则关注的台站列表内
String stations = rule.getStations(); String stations = rule.getStations();
if (!StrUtil.contains(stations, stationId)) if (!StrUtil.contains(stations, stationId))
@ -137,74 +146,71 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
info.setRuleId(rule.getId()); info.setRuleId(rule.getId());
info.setGroupId(rule.getContactGroup()); info.setGroupId(rule.getContactGroup());
info.setConditions(rule.getConditions()); info.setConditions(rule.getConditions());
judge(info,nuclidesCross); judge(info, nuclidesCross);
} }
} }
private void judge(Info info, Map<String,String> nuclidesCross){ private void judge(Info info, Map<String,String> nuclidesCross){
Set<String> nuclideNames = nuclidesCross.keySet(); Set<String> nuclideNames = nuclidesCross.keySet();
StringBuilder alarmInfo = new StringBuilder();
List<String> firstDetected;
List<NuclideInfo> moreThanAvg = new ArrayList<>();
String conditionStr = info.getConditions(); String conditionStr = info.getConditions();
String betaOrGamma = info.getBetaOrGamma(); String betaOrGamma = info.getBetaOrGamma();
String datasource = info.getDatasource(); String datasource = info.getDatasource();
String stationId = info.getStationId();
// 获取谱文件采样日期 如果为null 则默认为LocalDate.now()
LocalDate collDate = ObjectUtil.isNull(info.getCollectionDate()) ? LocalDate.now() :
info.getCollectionDate().toLocalDate();
List<String> conditions = ListUtil.toList(conditionStr.split(COMMA)); List<String> conditions = ListUtil.toList(conditionStr.split(COMMA));
List<String> firstDetected = new ArrayList<>(); // 首次发现
List<NuclideInfo> moreThanAvg = new ArrayList<>(); // 超浓度均值
List<String> meanwhile = new ArrayList<>(); // 同时出现两种及以上核素
for (String con : conditions) { for (String con : conditions) {
Condition condition = Condition.valueOf1(con); Condition condition = Condition.valueOf1(con);
if (ObjectUtil.isNotNull(condition)){ if (ObjectUtil.isNull(condition)) continue;
switch (condition){ switch (condition){
case FIRST_FOUND: // 首次发现该元素 case FIRST_FOUND: // 首次发现该元素
firstDetected = firstDetected(betaOrGamma,datasource,nuclideNames); firstDetected = firstDetected(betaOrGamma, datasource, nuclideNames);
if (CollUtil.isNotEmpty(firstDetected)){ break;
String message = "First discovery of nuclides: [" + StrUtil.join(COMMA,firstDetected) + "]"; case ABOVE_AVERAGE: // 元素浓度高于均值
alarmInfo.append(message); moreThanAvg = moreThanAvg(datasource, stationId, collDate, nuclidesCross);
} break;
break; case MEANWHILE: // 同时出现两种及以上核素
case ABOVE_AVERAGE: // 元素浓度高于均值 if (CollUtil.isNotEmpty(nuclideNames) && nuclideNames.size() >= 2)
moreThanAvg = moreThanAvg(datasource,nuclidesCross); meanwhile.addAll(nuclideNames);
if (CollUtil.isNotEmpty(moreThanAvg)){ break;
for (NuclideInfo nuclideInfo : moreThanAvg) { default:
String nuclide = nuclideInfo.getNuclide(); break;
String threshold = nuclideInfo.getThreshold();
String message = "Nuclide " + nuclide + "is above average: " + threshold;
alarmInfo.append(COMMA).append(message);
}
}
break;
case MEANWHILE: // 同时出现两种及以上核素
if (nuclideNames.size() >= 2){
String message = "Simultaneously detecting nuclides: [" + StrUtil.join(COMMA,nuclideNames) + "]";
alarmInfo.append(COMMA).append(message);
}
break;
default:
break;
}
} }
} }
if (StrUtil.isNotBlank(alarmInfo.toString())){ // 构建预警信息
// 保存报警日志 DataTool dataTool = DataTool.getInstance();
AlarmAnalysisLog logInfo = new AlarmAnalysisLog(); if (CollUtil.isNotEmpty(firstDetected))
BeanUtil.copyProperties(info,logInfo); dataTool.put("firstDetected", CollUtil.join(firstDetected, StrUtil.COMMA + StrUtil.SPACE));
SampleType sampleType = SampleType.typeOf(betaOrGamma); if (CollUtil.isNotEmpty(moreThanAvg)){
if (ObjectUtil.isNotNull(sampleType)) String above = moreThanAvg.stream()
logInfo.setSampleType(sampleType.getValue()); .map(item -> item.getNuclide() + "(" + item.getValue() + ")" + " > " + item.getThreshold())
if (alarmInfo.toString().startsWith(COMMA)) .collect(Collectors.joining(StrUtil.COMMA + StrUtil.SPACE));
alarmInfo = new StringBuilder(StrUtil.sub(alarmInfo.toString(), 1, alarmInfo.length())); dataTool.put("moreThanAvg", above);
logInfo.setAlarmInfo(alarmInfo.toString());
if (CollUtil.isNotEmpty(moreThanAvg))
logInfo.setNuclideInfoList(moreThanAvg);
logService.saveLog(logInfo);
// 发送报警信息
String groupId = info.getGroupId();
MessageDTO messageDTO = new MessageDTO();
messageDTO.setTitle("Nuclied Warn Info").setContent(alarmInfo.toString());
if (StrUtil.isNotBlank(groupId)) {
systemClient.sendMessage(messageDTO, groupId, ALL.getValue());
systemClient.pushMessageToSingle(messageDTO, groupId);
}
} }
if (CollUtil.isNotEmpty(meanwhile))
dataTool.put("meanwhile", CollUtil.join(meanwhile, StrUtil.COMMA + StrUtil.SPACE));
// 如果报警数据为空 则不需要发送报警信息和生成报警日志
if (MapUtil.isEmpty(dataTool.get())) return;
MessageDTO messageDTO = TemplateUtil.parse1(ANALYSIS_NUCLIDE.getCode(), dataTool.get());
// 保存报警日志
AlarmAnalysisLog logInfo = new AlarmAnalysisLog();
BeanUtil.copyProperties(info, logInfo);
SampleType sampleType = SampleType.typeOf(betaOrGamma);
if (ObjectUtil.isNotNull(sampleType))
logInfo.setSampleType(sampleType.getValue());
logInfo.setAlarmInfo(messageDTO.getContent());
if (CollUtil.isNotEmpty(moreThanAvg))
logInfo.setNuclideInfoList(moreThanAvg);
logService.saveLog(logInfo);
// 发送报警信息
String groupId = info.getGroupId();
systemClient.sendMessage(messageDTO, groupId, ALL.getValue());
systemClient.pushMessageToSingle(messageDTO, groupId);
} }
/** /**
@ -222,12 +228,12 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
/** /**
* 核素值大于历史浓度均值 * 核素值大于历史浓度均值
*/ */
private List<NuclideInfo> moreThanAvg(String dataSourceType, private List<NuclideInfo> moreThanAvg(String dataSourceType, String stationId,
Map<String,String> nuclidesCross){ LocalDate collDate, Map<String,String> nuclidesCross){
List<NuclideInfo> nuclideInfos = new ArrayList<>(); List<NuclideInfo> nuclideInfos = new ArrayList<>();
Set<String> nuclideNames = nuclidesCross.keySet(); Set<String> nuclideNames = nuclidesCross.keySet();
Map<String, String> nuclideAvgs = nuclideAvgService Map<String, String> nuclideAvgs = nuclideAvgService
.list(nuclideNames, dataSourceType).stream() .list(nuclideNames, dataSourceType, stationId, collDate).stream()
.collect(Collectors.toMap(AlarmAnalysisNuclideAvg::getNuclide, .collect(Collectors.toMap(AlarmAnalysisNuclideAvg::getNuclide,
AlarmAnalysisNuclideAvg::getVal)); AlarmAnalysisNuclideAvg::getVal));
for (Map.Entry<String, String> nuclide : nuclidesCross.entrySet()) { for (Map.Entry<String, String> nuclide : nuclidesCross.entrySet()) {
@ -246,41 +252,27 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
NuclideInfo nuclideInfo = new NuclideInfo(); NuclideInfo nuclideInfo = new NuclideInfo();
nuclideInfo.setNuclide(nuclideName); nuclideInfo.setNuclide(nuclideName);
nuclideInfo.setThreshold(avg.toString()); nuclideInfo.setThreshold(avg.toString());
nuclideInfo.setDatasource(type(dataSourceType)); nuclideInfo.setDatasource(DSType.typeOf(dataSourceType));
nuclideInfo.setValue(conc.toString()); // 对浓度值保留五位小数
nuclideInfo.setValue(NumUtil.keepStr(concValue, 5));
nuclideInfos.add(nuclideInfo); nuclideInfos.add(nuclideInfo);
} }
return nuclideInfos; return nuclideInfos;
} }
private String type(String dataSourceType){
switch (dataSourceType){
case CommonConstant.ARMDARR:
return "ARMDARR";
case CommonConstant.ARMDRRR:
return "ARMDRRR";
case CommonConstant.IDCARR:
return "IDCARR";
case CommonConstant.IDCRRR:
return "IDCRRR";
default:
return null;
}
}
private void init() { private void init() {
// start:生成临时Token到线程中 // start 生成临时Token到线程中
UserTokenContext.setToken(getTempToken()); UserTokenContext.setToken(getTempToken());
systemClient = SpringContextUtils.getBean(SystemClient.class); systemClient = SpringContextUtils.getBean(SystemClient.class);
redisStreamUtil = SpringContextUtils.getBean(RedisStreamUtil.class); redisStreamUtil = SpringContextUtils.getBean(RedisStreamUtil.class);
logService = SpringContextUtils.getBean(IAlarmAnalysisLogService.class); logService = SpringContextUtils.getBean(IAlarmAnalysisLogService.class);
analysisResultService = SpringContextUtils.getBean(AnalysisResultService.class);
ruleService = SpringContextUtils.getBean(IAlarmAnalysisRuleService.class); ruleService = SpringContextUtils.getBean(IAlarmAnalysisRuleService.class);
analysisResultService = SpringContextUtils.getBean(AnalysisResultService.class);
nuclideAvgService = SpringContextUtils.getBean(IAlarmAnalysisNuclideAvgService.class); nuclideAvgService = SpringContextUtils.getBean(IAlarmAnalysisNuclideAvgService.class);
} }
private void destroy(){ private void destroy(){
// end:删除临时Token // end 删除临时Token
UserTokenContext.remove(); UserTokenContext.remove();
} }
} }

36
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/redisStream/RedisStreamConfig.java
View File

@ -34,10 +34,6 @@ public class RedisStreamConfig {
// 每次轮询取几条消息 // 每次轮询取几条消息
private final Integer maxMsg = 10; private final Integer maxMsg = 10;
private final String alarmKey = RedisConstant.STREAM_ALARM;
private final String alarmGroup = RedisConstant.GROUP_ALARM;
private final String alarmConsumer = RedisConstant.CONSUMER_ALARM;
private final String analysisKey = RedisConstant.STREAM_ANALYSIS; private final String analysisKey = RedisConstant.STREAM_ANALYSIS;
private final String analysisGroup = RedisConstant.GROUP_ANALYSIS; private final String analysisGroup = RedisConstant.GROUP_ANALYSIS;
private final String analysisConsumer = RedisConstant.CONSUMER_ANALYSIS; private final String analysisConsumer = RedisConstant.CONSUMER_ANALYSIS;
@ -103,7 +99,7 @@ public class RedisStreamConfig {
// 独立消费 // 独立消费
/*streamMessageListenerContainer.receive(StreamOffset.fromStart(streamKey), /*streamMessageListenerContainer.receive(StreamOffset.fromStart(streamKey),
new ConsumeListener("独立消费", null, null));*/ new ConsumeListener("独立消费", null, null));*/
// 非独立消费
/* /*
register:用于注册一个消息监听器,该监听器将在每次有新的消息到达Redis Stream时被调用 register:用于注册一个消息监听器,该监听器将在每次有新的消息到达Redis Stream时被调用
这种方式适用于长期运行的消息消费者，它会持续监听Redis Stream并处理新到达的消息 这种方式适用于长期运行的消息消费者，它会持续监听Redis Stream并处理新到达的消息
@ -111,24 +107,28 @@ public class RedisStreamConfig {
receive:用于主动从Redis Stream中获取消息并进行处理,你可以指定要获取的消息数量和超时时间 receive:用于主动从Redis Stream中获取消息并进行处理,你可以指定要获取的消息数量和超时时间
这种方式适用于需要主动控制消息获取的场景,例如批量处理消息或定时任务 这种方式适用于需要主动控制消息获取的场景,例如批量处理消息或定时任务
**/ **/
// 注册消费组A中的消费者A1,手动ACK /* 1.需要手动确认消费消息 */
/*ConsumerStreamReadRequest<String> readA1 = StreamMessageListenerContainer // 1.1 使用 register 方式
StreamMessageListenerContainer.ConsumerStreamReadRequest<String> readRequest = StreamMessageListenerContainer
.StreamReadRequest .StreamReadRequest
.builder(StreamOffset.create(warnKey, ReadOffset.lastConsumed())) .builder(StreamOffset.create(analysisKey, ReadOffset.lastConsumed()))
.consumer(Consumer.from(groupWarnA, consumerWarnA1)) .consumer(Consumer.from(analysisGroup, analysisConsumer))
// 手动确认消费了消息 默认为自动确认消息
.autoAcknowledge(false) .autoAcknowledge(false)
// 如果消费者发生了异常,是否禁止消费者消费 // 如果消费者发生了异常 不禁止消费者消费 默认为禁止
.cancelOnError(throwable -> false) .cancelOnError(throwable -> false)
.build(); .build();
ConsumeA1 consumeA1 = new ConsumeA1(groupWarnA, consumerWarnA1); AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup, analysisConsumer);
streamMessageListenerContainer.register(readA1, consumeA1);*/ streamMessageListenerContainer.register(readRequest, analysis);
AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup,analysisConsumer); // 1.2 使用 receive 方式
/*AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup, analysisConsumer);
streamMessageListenerContainer.receive(Consumer.from(analysisGroup, analysisConsumer), streamMessageListenerContainer.receive(Consumer.from(analysisGroup, analysisConsumer),
StreamOffset.create(analysisKey, ReadOffset.lastConsumed()), analysis); StreamOffset.create(analysisKey, ReadOffset.lastConsumed()), analysis);*/
// 创建消费组A中的消费者A2,自动ACK /* 2.自动确认消费消息 */
/* ConsumeA2 consumeA2 = new ConsumeA2(consumerWarnA2); // 2.1 使用 receive 方式
streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupWarnA, consumerWarnA2), /*AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup,analysisConsumer);
StreamOffset.create(warnKey, ReadOffset.lastConsumed()), consumeA2);*/ streamMessageListenerContainer.receiveAutoAck(Consumer.from(analysisGroup, analysisConsumer),
StreamOffset.create(analysisKey, ReadOffset.lastConsumed()), analysis);*/
return streamMessageListenerContainer; return streamMessageListenerContainer;
} }
} }

4
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/IAlarmAnalysisNuclideAvgService.java
View File

@ -6,12 +6,14 @@ import org.jeecg.modules.base.entity.postgre.AlarmAnalysisNuclideAvg;
import com.baomidou.mybatisplus.extension.service.IService; import com.baomidou.mybatisplus.extension.service.IService;
import org.jeecg.modules.base.bizVo.NuclideAvgVo; import org.jeecg.modules.base.bizVo.NuclideAvgVo;
import java.time.LocalDate;
import java.util.List; import java.util.List;
import java.util.Set; import java.util.Set;
public interface IAlarmAnalysisNuclideAvgService extends IService<AlarmAnalysisNuclideAvg> { public interface IAlarmAnalysisNuclideAvgService extends IService<AlarmAnalysisNuclideAvg> {
List<AlarmAnalysisNuclideAvg> list(Set<String> nuclideNames,String dataSourceType); List<AlarmAnalysisNuclideAvg> list(Set<String> nuclideNames, String dataSourceType,
String stationId, LocalDate collDate);
Page<NuclideAvgDto> findPage(NuclideAvgVo nuclideAvgVo); Page<NuclideAvgDto> findPage(NuclideAvgVo nuclideAvgVo);
} }

1
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/AlarmAnalysisLogServiceImpl.java
View File

@ -75,6 +75,7 @@ public class AlarmAnalysisLogServiceImpl extends ServiceImpl<AlarmAnalysisLogMap
try { try {
ObjectMapper mapper = new ObjectMapper(); ObjectMapper mapper = new ObjectMapper();
List<NuclideInfo> nuclideInfos = mapper.readValue(nuclideInfo, new TypeReference<List<NuclideInfo>>() {}); List<NuclideInfo> nuclideInfos = mapper.readValue(nuclideInfo, new TypeReference<List<NuclideInfo>>() {});
nuclideInfos.forEach(NuclideInfo::keepSix);
logDto.setNuclideList(nuclideInfos); logDto.setNuclideList(nuclideInfos);
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
log.error("NuclideInfo解析异常: {}", e.getMessage()); log.error("NuclideInfo解析异常: {}", e.getMessage());

13
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/AlarmAnalysisNuclideAvgServiceImpl.java
View File

@ -15,6 +15,7 @@ import org.springframework.stereotype.Service;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import java.time.LocalDate; import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
@ -27,13 +28,15 @@ public class AlarmAnalysisNuclideAvgServiceImpl extends ServiceImpl<AlarmAnalysi
private SystemClient systemClient; private SystemClient systemClient;
@Override @Override
public List<AlarmAnalysisNuclideAvg> list(Set<String> nuclideNames,String dataSourceType) { public List<AlarmAnalysisNuclideAvg> list(Set<String> nuclideNames, String dataSourceType,
LocalDate dayAgo = LocalDate.now().minusDays(1); String stationId, LocalDate collDate) {
LocalDate dayAgo = collDate.minusDays(1);
LambdaQueryWrapper<AlarmAnalysisNuclideAvg> wrapper = new LambdaQueryWrapper<>(); LambdaQueryWrapper<AlarmAnalysisNuclideAvg> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(AlarmAnalysisNuclideAvg::getDataSourceType,dataSourceType); wrapper.eq(AlarmAnalysisNuclideAvg::getStationId, stationId);
wrapper.eq(AlarmAnalysisNuclideAvg::getCaclDate,dayAgo); wrapper.eq(AlarmAnalysisNuclideAvg::getDataSourceType, dataSourceType);
wrapper.eq(AlarmAnalysisNuclideAvg::getCaclDate, dayAgo);
wrapper.in(AlarmAnalysisNuclideAvg::getNuclide,nuclideNames); wrapper.in(AlarmAnalysisNuclideAvg::getNuclide,nuclideNames);
return list(wrapper); return this.list(wrapper);
} }
@Override @Override

34
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/feignclient/ManageUtil.java
View File

@ -1,5 +1,6 @@
package org.jeecg.modules.feignclient; package org.jeecg.modules.feignclient;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.vo.Result; import org.jeecg.common.api.vo.Result;
import org.jeecg.common.constant.RedisConstant; import org.jeecg.common.constant.RedisConstant;
import org.jeecg.common.system.util.JwtUtil; import org.jeecg.common.system.util.JwtUtil;
@ -9,6 +10,7 @@ import org.jeecg.modules.base.dto.LoginResult;
import org.jeecg.modules.base.dto.LoginVo; import org.jeecg.modules.base.dto.LoginVo;
import org.springframework.core.env.Environment; import org.springframework.core.env.Environment;
@Slf4j
public class ManageUtil { public class ManageUtil {
private static RedisUtil redisUtil; private static RedisUtil redisUtil;
@ -31,20 +33,28 @@ public class ManageUtil {
* 登录运管系统 获取Token * 登录运管系统 获取Token
* */ * */
public static String getToken(){ public static String getToken(){
if (redisUtil.hasKey(RedisConstant.MANAGE_TOKEN)) String token = null;
return (String) redisUtil.get(RedisConstant.MANAGE_TOKEN); try {
if (redisUtil.hasKey(RedisConstant.MANAGE_TOKEN))
LoginVo loginVo = new LoginVo(username, password); return (String) redisUtil.get(RedisConstant.MANAGE_TOKEN);
Result<LoginResult> loginRes = monitorSystem.login(loginVo); LoginVo loginVo = new LoginVo(username, password);
String token = loginRes.getResult().getToken(); Result<LoginResult> loginRes = monitorSystem.login(loginVo);
redisUtil.set(RedisConstant.MANAGE_TOKEN, token, JwtUtil.EXPIRE_TIME / 1000 - 10); token = loginRes.getResult().getToken();
return token; redisUtil.set(RedisConstant.MANAGE_TOKEN, token, JwtUtil.EXPIRE_TIME / 1000 - 10);
return token;
}catch (RuntimeException e){
return token;
}
} }
public static void refreshToken(){ public static void refreshToken(){
LoginVo loginVo = new LoginVo(username, password); try {
Result<LoginResult> loginRes = monitorSystem.login(loginVo); LoginVo loginVo = new LoginVo(username, password);
String token = loginRes.getResult().getToken(); Result<LoginResult> loginRes = monitorSystem.login(loginVo);
redisUtil.set(RedisConstant.MANAGE_TOKEN, token, JwtUtil.EXPIRE_TIME / 1000 - 10); String token = loginRes.getResult().getToken();
redisUtil.set(RedisConstant.MANAGE_TOKEN, token, JwtUtil.EXPIRE_TIME / 1000 - 10);
}catch (RuntimeException e){
log.error("运管系统登录异常, Token刷新失败: {}", e.getMessage());
}
} }
} }

50
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/SendMessage.java
View File

@ -4,6 +4,7 @@ import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.map.MapUtil; import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.dto.message.MessageDTO; import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.constant.SymbolConstant; import org.jeecg.common.constant.SymbolConstant;
import org.jeecg.common.constant.enums.MessageTypeEnum; import org.jeecg.common.constant.enums.MessageTypeEnum;
@ -27,6 +28,7 @@ import static org.jeecg.common.constant.enums.MessageTypeEnum.*;
* @author nieziyan * @author nieziyan
* @date 2023-06-21 * @date 2023-06-21
*/ */
@Slf4j
@Component @Component
public class SendMessage { public class SendMessage {
@ -50,6 +52,7 @@ public class SendMessage {
*/ */
public void send(MessageDTO messageDTO, String groupId, String notific){ public void send(MessageDTO messageDTO, String groupId, String notific){
Map<String, String> contact = getContact(groupId); Map<String, String> contact = getContact(groupId);
if (MapUtil.isEmpty(contact)) return;
if (StrUtil.isBlank(notific)) return; if (StrUtil.isBlank(notific)) return;
List<String> ways = ListUtil.toList(StrUtil.split(notific, COMMA)); List<String> ways = ListUtil.toList(StrUtil.split(notific, COMMA));
if (ways.contains(ALL.getValue())) if (ways.contains(ALL.getValue()))
@ -96,26 +99,33 @@ public class SendMessage {
* @param groupId * @param groupId
*/ */
private Map<String, String> getContact(String groupId){ private Map<String, String> getContact(String groupId){
List<String> userIds = abnormalAlarmClient.userIds(groupId).getResult();
// 查询用户信息
List<SysUser> sysUsers = sysUserService.listByIds(userIds);
// 用户名
String usernameList = sysUsers.stream().map(SysUser::getUsername).filter(StrUtil::isNotBlank)
.collect(Collectors.joining(COMMA));
// 邮箱
String emailList = sysUsers.stream().map(SysUser::getEmail).filter(StrUtil::isNotBlank)
.collect(Collectors.joining(COMMA));
// 用户名-邮箱Map
Map<String, String> userEmail = sysUsers.stream()
.collect(Collectors.toMap(SysUser::getUsername, SysUser::getEmail));
// 手机号码
String phoneList = sysUsers.stream().map(SysUser::getPhone).filter(StrUtil::isNotBlank)
.collect(Collectors.joining(COMMA));
Map<String,String> result = new HashMap<>(); Map<String,String> result = new HashMap<>();
result.put(SYSTEM, usernameList); try {
result.put(EMAIL, emailList); List<String> userIds = abnormalAlarmClient.userIds(groupId).getResult();
result.put(SMS, phoneList); // 查询用户信息
result.putAll(userEmail); List<SysUser> sysUsers = sysUserService.listByIds(userIds);
return result; // 用户名
String usernameList = sysUsers.stream().map(SysUser::getUsername).filter(StrUtil::isNotBlank)
.collect(Collectors.joining(COMMA));
// 邮箱
String emailList = sysUsers.stream().map(SysUser::getEmail).filter(StrUtil::isNotBlank)
.collect(Collectors.joining(COMMA));
// 用户名-邮箱Map
Map<String, String> userEmail = sysUsers.stream()
.filter(item -> StrUtil.isNotBlank(item.getEmail()))
.collect(Collectors.toMap(SysUser::getUsername, SysUser::getEmail));
// 手机号码
String phoneList = sysUsers.stream().map(SysUser::getPhone).filter(StrUtil::isNotBlank)
.collect(Collectors.joining(COMMA));
result.put(SYSTEM, usernameList);
result.put(EMAIL, emailList);
result.put(SMS, phoneList);
result.putAll(userEmail);
return result;
}catch (Exception e) {
log.error("获取收件人联系信息异常: {}", e.getMessage());
return result;
}
} }
} }

10
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/job/DatabaseJob.java
View File

@ -6,6 +6,7 @@ import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import feign.FeignException;
import lombok.Data; import lombok.Data;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.dto.message.MessageDTO; import org.jeecg.common.api.dto.message.MessageDTO;
@ -102,6 +103,10 @@ public class DatabaseJob extends Monitor implements Job{
}*/ }*/
// 向运管查询监控项数据 // 向运管查询监控项数据
String token = ManageUtil.getToken(); String token = ManageUtil.getToken();
if(StrUtil.isBlank(token)){
log.error("运管系统登录异常, Token获取失败");
continue;
}
Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token); Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token);
ItemHistory itemHistory = result.getResult(); ItemHistory itemHistory = result.getResult();
if (ObjectUtil.isNull(itemHistory)){ if (ObjectUtil.isNull(itemHistory)){
@ -141,9 +146,12 @@ public class DatabaseJob extends Monitor implements Job{
getSendMessage().send(messageDTO, groupId, notific); getSendMessage().send(messageDTO, groupId, notific);
getPushAppUtil().pushToSingle(messageDTO, groupId); getPushAppUtil().pushToSingle(messageDTO, groupId);
} }
} catch (FeignException.Unauthorized e){
ManageUtil.refreshToken();
log.warn("向运管系统查询ItemHistory信息异常: Token失效,已刷新Token");
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
log.error("Database预警规则: {}解析失败,失败原因: {}", operator, e.getMessage()); log.error("Database预警规则: {}解析失败,失败原因: {}", operator, e.getMessage());
}catch (Exception e){ } catch (Exception e){
log.error("Database监控异常: {}", e.getMessage()); log.error("Database监控异常: {}", e.getMessage());
} }
} }

8
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/job/ServerJob.java
View File

@ -85,6 +85,10 @@ public class ServerJob extends Monitor implements Job {
// 向运管查询监控项数据 // 向运管查询监控项数据
String token = ManageUtil.getToken(); String token = ManageUtil.getToken();
if(StrUtil.isBlank(token)){
log.error("运管系统登录异常, Token获取失败");
continue;
}
Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token); Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token);
ItemHistory itemHistory = result.getResult(); ItemHistory itemHistory = result.getResult();
if (ObjectUtil.isNull(itemHistory)){ if (ObjectUtil.isNull(itemHistory)){
@ -124,12 +128,12 @@ public class ServerJob extends Monitor implements Job {
getSendMessage().send(messageDTO, groupId, notific); getSendMessage().send(messageDTO, groupId, notific);
getPushAppUtil().pushToSingle(messageDTO, groupId); getPushAppUtil().pushToSingle(messageDTO, groupId);
} }
}catch (FeignException.Unauthorized e){ } catch (FeignException.Unauthorized e){
ManageUtil.refreshToken(); ManageUtil.refreshToken();
log.warn("向运管系统查询ItemHistory信息异常: Token失效,已刷新Token"); log.warn("向运管系统查询ItemHistory信息异常: Token失效,已刷新Token");
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
log.error("Server预警规则: {}解析失败,失败原因: {}", operator, e.getMessage()); log.error("Server预警规则: {}解析失败,失败原因: {}", operator, e.getMessage());
}catch (Exception e){ } catch (Exception e){
log.error("Server监控异常: {}", e.getMessage()); log.error("Server监控异常: {}", e.getMessage());
} }
} }

9
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/jobs/DatabaseJob.java
View File

@ -6,6 +6,7 @@ import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import feign.FeignException;
import lombok.Data; import lombok.Data;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.dto.message.MessageDTO; import org.jeecg.common.api.dto.message.MessageDTO;
@ -86,6 +87,10 @@ public class DatabaseJob extends Monitor {
// 向运管查询监控项数据 // 向运管查询监控项数据
String token = ManageUtil.getToken(); String token = ManageUtil.getToken();
if(StrUtil.isBlank(token)){
log.error("运管系统登录异常, Token获取失败");
continue;
}
Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token); Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token);
ItemHistory itemHistory = result.getResult(); ItemHistory itemHistory = result.getResult();
if (ObjectUtil.isNull(itemHistory)){ if (ObjectUtil.isNull(itemHistory)){
@ -125,11 +130,13 @@ public class DatabaseJob extends Monitor {
getSendMessage().send(messageDTO, groupId, notific); getSendMessage().send(messageDTO, groupId, notific);
getPushAppUtil().pushToSingle(messageDTO, groupId); getPushAppUtil().pushToSingle(messageDTO, groupId);
} }
}catch (FeignException.Unauthorized e){
ManageUtil.refreshToken();
log.warn("向运管系统查询ItemHistory信息异常: Token失效,已刷新Token");
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
log.error("Database预警规则: {}解析失败,失败原因: {}", operator, e.getMessage()); log.error("Database预警规则: {}解析失败,失败原因: {}", operator, e.getMessage());
}catch (Exception e){ }catch (Exception e){
log.error("Database监控异常: {}", e.getMessage()); log.error("Database监控异常: {}", e.getMessage());
e.printStackTrace();
} }
} }
destroy(); destroy();

1
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/jobs/EmailJob.java
View File

@ -111,7 +111,6 @@ public class EmailJob extends Monitor{
log.error("Email预警规则: {}解析失败,失败原因: {}", operator, e.getMessage()); log.error("Email预警规则: {}解析失败,失败原因: {}", operator, e.getMessage());
}catch (Exception e){ }catch (Exception e){
log.error("Email监控异常: {}", e.getMessage()); log.error("Email监控异常: {}", e.getMessage());
e.printStackTrace();
} }
} }
destroy(); destroy();

9
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/jobs/ServerJob.java
View File

@ -84,6 +84,10 @@ public class ServerJob extends Monitor{
// 向运管查询监控项数据 // 向运管查询监控项数据
String token = ManageUtil.getToken(); String token = ManageUtil.getToken();
if(StrUtil.isBlank(token)){
log.error("运管系统登录异常, Token获取失败");
continue;
}
Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token); Result<ItemHistory> result = getMonitorSystem().itemBack(itemId, itemType, start, end, token);
ItemHistory itemHistory = result.getResult(); ItemHistory itemHistory = result.getResult();
if (ObjectUtil.isNull(itemHistory)){ if (ObjectUtil.isNull(itemHistory)){
@ -123,14 +127,13 @@ public class ServerJob extends Monitor{
getSendMessage().send(messageDTO, groupId, notific); getSendMessage().send(messageDTO, groupId, notific);
getPushAppUtil().pushToSingle(messageDTO, groupId); getPushAppUtil().pushToSingle(messageDTO, groupId);
} }
}catch (FeignException.Unauthorized e){ } catch (FeignException.Unauthorized e){
ManageUtil.refreshToken(); ManageUtil.refreshToken();
log.warn("向运管系统查询ItemHistory信息异常: Token失效,已刷新Token"); log.warn("向运管系统查询ItemHistory信息异常: Token失效,已刷新Token");
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
log.error("Server预警规则: {}解析失败,失败原因: {}", operator, e.getMessage()); log.error("Server预警规则: {}解析失败,失败原因: {}", operator, e.getMessage());
}catch (Exception e){ } catch (Exception e){
log.error("Server监控异常: {}", e.getMessage()); log.error("Server监控异常: {}", e.getMessage());
e.printStackTrace();
} }
} }
destroy(); destroy();

3
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/jobs/TableSpaceJob.java
View File

@ -44,7 +44,7 @@ public class TableSpaceJob extends Monitor {
/** /**
* 解析Oracle 表空间预警规则 * 解析Oracle 表空间预警规则
**/ **/
@Scheduled(cron = "${task.period-space:0 0 */1 * * ?}") @Scheduled(cron = "${task.period-space:0 0/1 * * * ?}")
public void execute(){ public void execute(){
init(); init();
@ -124,7 +124,6 @@ public class TableSpaceJob extends Monitor {
log.error("Database-TableSpace预警规则: {}解析失败,失败原因: {}", operator, e.getMessage()); log.error("Database-TableSpace预警规则: {}解析失败,失败原因: {}", operator, e.getMessage());
}catch (Exception e){ }catch (Exception e){
log.error("Database-TableSpace监控异常: {}", e.getMessage()); log.error("Database-TableSpace监控异常: {}", e.getMessage());
e.printStackTrace();
} }
} }
destroy(); destroy();