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

fix：消息队列消费模式修改

Browse Source
This commit is contained in:
nieziyan 2024-04-15 18:37:10 +08:00
parent d5f40c9023
commit 22b31b5daf
6 changed files with 67 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/RedisConstant.java
View File

@ -13,16 +13,10 @@ public interface RedisConstant {
*/
String PREFIX_SILENCE = "SilenceCycle:";
String STREAM_ALARM = "Stream:Alarm";
String STREAM_ANALYSIS = "Stream:Analysis";
String GROUP_ALARM = "Group_Alarm";
String GROUP_ANALYSIS = "Group_Analysis";
String CONSUMER_ALARM = "Consumer_Alarm";
String CONSUMER_ANALYSIS = "Consumer_Analysis";
String NUCLIDE_LINES_LIB = "Nuclide_Lines_Lib:";

2
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/enums/Condition.java
View File

@ -9,7 +9,7 @@ import lombok.Getter;
public enum Condition {
FIRST_FOUND("1"), ABOVE_AVERAGE("2"), MEANWHILE("3");
private String value;
private final String value;
public static Condition valueOf1(String value){
for (Condition condition : Condition.values()) {

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

@ -18,10 +18,8 @@ import org.jeecg.common.config.mqtoken.UserTokenContext;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.constant.SymbolConstant;
import org.jeecg.common.constant.enums.SampleType;
import org.jeecg.common.util.DataTool;
import org.jeecg.common.util.RedisStreamUtil;
import org.jeecg.common.util.SpringContextUtils;
import org.jeecg.common.util.TemplateUtil;
import org.jeecg.common.util.*;
import org.jeecg.common.util.dynamic.db.FreemarkerParseFactory;
import org.jeecg.modules.base.dto.NuclideInfo;
import org.jeecg.modules.base.dto.Info;
import org.jeecg.modules.base.entity.postgre.AlarmAnalysisLog;
@ -45,6 +43,7 @@ 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.time.LocalDate;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@ -80,7 +79,7 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
try {
String streamKey = message.getStream();
init();
/**
/*
* 新消息在未进行ACK之前,状态也为pending,
* 直接消费所有异常未确认的消息和新消息
*/
@ -95,10 +94,10 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
// 消费完成后,手动确认消费消息[消息消费成功]
// 否则就是消费抛出异常,进入pending_ids[消息消费失败]
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());
}finally {
destroy();
@ -118,7 +117,8 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
for (AlarmAnalysisRule rule : rules) {
// 当前规则是否有报警条件
String conditionStr = rule.getConditions();
if (StrUtil.isBlank(conditionStr)) continue;
if (StrUtil.isBlank(conditionStr))
continue;
// 是否在当前规则关注的台站列表内
String stations = rule.getStations();
if (!StrUtil.contains(stations, stationId))
@ -146,7 +146,7 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
info.setRuleId(rule.getId());
info.setGroupId(rule.getContactGroup());
info.setConditions(rule.getConditions());
judge(info,nuclidesCross);
judge(info, nuclidesCross);
}
}
@ -155,27 +155,31 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
String conditionStr = info.getConditions();
String betaOrGamma = info.getBetaOrGamma();
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> firstDetected = new ArrayList<>(); // 首次发现
List<NuclideInfo> moreThanAvg = new ArrayList<>(); // 超浓度均值
List<String> meanwhile = new ArrayList<>(); // 同时出现两种及以上核素
for (String con : conditions) {
Condition condition = Condition.valueOf1(con);
if (ObjectUtil.isNotNull(condition)){
switch (condition){
case FIRST_FOUND: // 首次发现该元素
firstDetected = firstDetected(betaOrGamma, datasource, nuclideNames);
break;
case ABOVE_AVERAGE: // 元素浓度高于均值
moreThanAvg = moreThanAvg(datasource,nuclidesCross);
break;
case MEANWHILE: // 同时出现两种及以上核素
if (CollUtil.isNotEmpty(nuclideNames) && nuclideNames.size() >= 2)
meanwhile.addAll(nuclideNames);
break;
default:
break;
}
if (ObjectUtil.isNull(condition)) continue;
switch (condition){
case FIRST_FOUND: // 首次发现该元素
firstDetected = firstDetected(betaOrGamma, datasource, nuclideNames);
break;
case ABOVE_AVERAGE: // 元素浓度高于均值
moreThanAvg = moreThanAvg(datasource, stationId, collDate, nuclidesCross);
break;
case MEANWHILE: // 同时出现两种及以上核素
if (CollUtil.isNotEmpty(nuclideNames) && nuclideNames.size() >= 2)
meanwhile.addAll(nuclideNames);
break;
default:
break;
}
}
// 构建预警信息
@ -224,12 +228,12 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
/**
* 核素值大于历史浓度均值
*/
private List<NuclideInfo> moreThanAvg(String dataSourceType,
Map<String,String> nuclidesCross){
private List<NuclideInfo> moreThanAvg(String dataSourceType, String stationId,
LocalDate collDate, Map<String,String> nuclidesCross){
List<NuclideInfo> nuclideInfos = new ArrayList<>();
Set<String> nuclideNames = nuclidesCross.keySet();
Map<String, String> nuclideAvgs = nuclideAvgService
.list(nuclideNames, dataSourceType).stream()
.list(nuclideNames, dataSourceType, stationId, collDate).stream()
.collect(Collectors.toMap(AlarmAnalysisNuclideAvg::getNuclide,
AlarmAnalysisNuclideAvg::getVal));
for (Map.Entry<String, String> nuclide : nuclidesCross.entrySet()) {
@ -249,25 +253,26 @@ public class AnalysisConsumer implements StreamListener<String, ObjectRecord<Str
nuclideInfo.setNuclide(nuclideName);
nuclideInfo.setThreshold(avg.toString());
nuclideInfo.setDatasource(DSType.typeOf(dataSourceType));
nuclideInfo.setValue(conc.toString());
// 对浓度值保留五位小数
nuclideInfo.setValue(NumUtil.keepStr(concValue, 5));
nuclideInfos.add(nuclideInfo);
}
return nuclideInfos;
}
private void init() {
// start:生成临时Token到线程中
// start 生成临时Token到线程中
UserTokenContext.setToken(getTempToken());
systemClient = SpringContextUtils.getBean(SystemClient.class);
redisStreamUtil = SpringContextUtils.getBean(RedisStreamUtil.class);
logService = SpringContextUtils.getBean(IAlarmAnalysisLogService.class);
analysisResultService = SpringContextUtils.getBean(AnalysisResultService.class);
ruleService = SpringContextUtils.getBean(IAlarmAnalysisRuleService.class);
analysisResultService = SpringContextUtils.getBean(AnalysisResultService.class);
nuclideAvgService = SpringContextUtils.getBean(IAlarmAnalysisNuclideAvgService.class);
}
private void destroy(){
// end:删除临时Token
// end 删除临时Token
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 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 analysisGroup = RedisConstant.GROUP_ANALYSIS;
private final String analysisConsumer = RedisConstant.CONSUMER_ANALYSIS;
@ -103,7 +99,7 @@ public class RedisStreamConfig {
// 独立消费
/*streamMessageListenerContainer.receive(StreamOffset.fromStart(streamKey),
new ConsumeListener("独立消费", null, null));*/
// 非独立消费
/*
register:用于注册一个消息监听器,该监听器将在每次有新的消息到达Redis Stream时被调用
这种方式适用于长期运行的消息消费者，它会持续监听Redis Stream并处理新到达的消息
@ -111,24 +107,28 @@ public class RedisStreamConfig {
receive:用于主动从Redis Stream中获取消息并进行处理,你可以指定要获取的消息数量和超时时间
这种方式适用于需要主动控制消息获取的场景,例如批量处理消息或定时任务
**/
// 注册消费组A中的消费者A1,手动ACK
/*ConsumerStreamReadRequest<String> readA1 = StreamMessageListenerContainer
/* 1.需要手动确认消费消息 */
// 1.1 使用 register 方式
StreamMessageListenerContainer.ConsumerStreamReadRequest<String> readRequest = StreamMessageListenerContainer
.StreamReadRequest
.builder(StreamOffset.create(warnKey, ReadOffset.lastConsumed()))
.consumer(Consumer.from(groupWarnA, consumerWarnA1))
.builder(StreamOffset.create(analysisKey, ReadOffset.lastConsumed()))
.consumer(Consumer.from(analysisGroup, analysisConsumer))
// 手动确认消费了消息 默认为自动确认消息
.autoAcknowledge(false)
// 如果消费者发生了异常,是否禁止消费者消费
// 如果消费者发生了异常 不禁止消费者消费 默认为禁止
.cancelOnError(throwable -> false)
.build();
ConsumeA1 consumeA1 = new ConsumeA1(groupWarnA, consumerWarnA1);
streamMessageListenerContainer.register(readA1, consumeA1);*/
AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup,analysisConsumer);
AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup, analysisConsumer);
streamMessageListenerContainer.register(readRequest, analysis);
// 1.2 使用 receive 方式
/*AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup, analysisConsumer);
streamMessageListenerContainer.receive(Consumer.from(analysisGroup, analysisConsumer),
StreamOffset.create(analysisKey, ReadOffset.lastConsumed()), analysis);
// 创建消费组A中的消费者A2,自动ACK
/* ConsumeA2 consumeA2 = new ConsumeA2(consumerWarnA2);
streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupWarnA, consumerWarnA2),
StreamOffset.create(warnKey, ReadOffset.lastConsumed()), consumeA2);*/
StreamOffset.create(analysisKey, ReadOffset.lastConsumed()), analysis);*/
/* 2.自动确认消费消息 */
// 2.1 使用 receive 方式
/*AnalysisConsumer analysis = new AnalysisConsumer(analysisGroup,analysisConsumer);
streamMessageListenerContainer.receiveAutoAck(Consumer.from(analysisGroup, analysisConsumer),
StreamOffset.create(analysisKey, ReadOffset.lastConsumed()), analysis);*/
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 org.jeecg.modules.base.bizVo.NuclideAvgVo;
import java.time.LocalDate;
import java.util.List;
import java.util.Set;
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);
}

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