Microservice/govern
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(harmonic): 重构事件通知服务并迁移至系统模块

Browse Source 
- 移除 AppNotificationService 和 SmsNotificationService 服务类
- 新增消息发送服务相关接口和实现类到系统模块
- 迁移 AsyncConfig 配置类到 cs-system 模块
- 新增事件用户关系表的 Mapper、Service 接口及实现类
- 更新 CsEventPO 添加 landPoint 字段用于存储事件落点信息
- 修改事件处理逻辑使用新的消息发送 Feign 客户端
- 集成事件原因分析功能并更新数据库字段映射
- 更新设备详情 DTO 添加 nDid 属性支持
- 优化事件通知和短信发送的消息内容格式
- 移除旧的设备消息客户端依赖并使用新服务接口
This commit is contained in:
xy
2026-05-27 11:12:32 +08:00
parent 202888ca14
commit f81be47e5f
19 changed files with 581 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/config/AsyncConfig.java
View File

@@ -1,42 +0,0 @@
package com.njcn.csharmonic.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
/**
* @author xy
*/
@Configuration
@Slf4j
public class AsyncConfig implements AsyncConfigurer {
@Bean("eventNotificationExecutor")
public Executor eventNotificationExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(20);
executor.setQueueCapacity(200);
executor.setThreadNamePrefix("event-notify-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
@Bean("smsNotificationExecutor")
public Executor smsNotificationExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("sms-notify-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}

3
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/mapper/mapping/CsEventUserPOMapper.xml
View File

@@ -168,7 +168,8 @@
b.type type,
b.LEVEL LEVEL,
b.location location,
c.dev_type devType
c.dev_type devType,
b.land_point landPoint
<if test="csEventUserQueryPage != null and csEventUserQueryPage.type != null and csEventUserQueryPage.type != '' and csEventUserQueryPage.type ==0">
,d.NAME lineName
</if>

113
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/AppNotificationService.java
View File

@@ -1,113 +0,0 @@
package com.njcn.csharmonic.service;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import com.njcn.access.pojo.dto.NoticeUserDto;
import com.njcn.access.utils.SendMessageUtil;
import com.njcn.csdevice.api.DeviceMessageFeignClient;
import com.njcn.csdevice.param.DeviceMessageParam;
import com.njcn.csdevice.pojo.dto.DevDetailDTO;
import com.njcn.csharmonic.pojo.po.CsEventUserPO;
import com.njcn.system.api.EpdFeignClient;
import com.njcn.user.pojo.po.User;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
/**
* @author xy
*/
@Service
@Slf4j
@RequiredArgsConstructor
public class AppNotificationService {
private final EpdFeignClient epdFeignClient;
private final SendMessageUtil sendMessageUtil;
private final DeviceMessageFeignClient deviceMessageClient;
private final CsEventUserPOService csEventUserPOService;
@Async("eventNotificationExecutor")
public void asyncSaveEventUserAndNotify(String eventId
, List<String> eventUser
, DevDetailDTO devDetailDto
, LocalDateTime time
, Integer eventType
, Double amplitude
, Double persistTime) {
NoticeUserDto noticeUserDto = new NoticeUserDto();
NoticeUserDto.Payload payload = new NoticeUserDto.Payload();
List<CsEventUserPO> result = new ArrayList<>();
eventUser.forEach(item -> {
CsEventUserPO csEventUser = new CsEventUserPO();
csEventUser.setUserId(item);
csEventUser.setStatus(0);
csEventUser.setEventId(eventId);
result.add(csEventUser);
});
DeviceMessageParam param = new DeviceMessageParam();
param.setUserList(eventUser);
param.setEventType(0);
List<User> users = deviceMessageClient.getSendUserByType(param).getData();
if (CollectionUtil.isNotEmpty(users)) {
List<String> devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
noticeUserDto.setPushClientId(devCodeList);
noticeUserDto.setTitle("暂态事件");
}
String eventName = epdFeignClient.findByName(getTag(eventType)).getData().getShowName();
String content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+ "于" + time.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂态事件，事件类型："
+ eventName
+ "，特征幅值：" + amplitude + "%"
+ "，持续时间：" + persistTime + "s";
noticeUserDto.setContent(content);
payload.setType(0);
payload.setPath("/pages/index/message1?type=" + payload.getType());
noticeUserDto.setPayload(payload);
if (CollectionUtil.isNotEmpty(noticeUserDto.getPushClientId())) {
List<String> filteredList = noticeUserDto.getPushClientId().stream()
.filter(s -> s != null && !s.isEmpty())
.distinct()
.collect(Collectors.toList());
if (CollectionUtil.isNotEmpty(filteredList)) {
noticeUserDto.setPushClientId(filteredList);
sendMessageUtil.sendEventToUser(noticeUserDto);
}
}
if (CollectionUtil.isNotEmpty(result)) {
csEventUserPOService.saveBatch(result);
}
}
public String getTag(Integer type) {
String tag;
switch (type) {
case 1:
tag = "Evt_Sys_DipStr";
break;
case 2:
tag = "Evt_Sys_SwlStr";
break;
case 3:
tag = "Evt_Sys_IntrStr";
break;
case 4:
tag = "Transient";
break;
default:
tag = "Un_Know";
break;
}
return tag;
}
}

60
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/SmsNotificationService.java
View File

@@ -1,60 +0,0 @@
package com.njcn.csharmonic.service;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.StrUtil;
import com.njcn.csdevice.api.SmsSendFeignClient;
import com.njcn.csdevice.pojo.dto.DevDetailDTO;
import com.njcn.cssystem.api.AppMsgSetFeignClient;
import com.njcn.user.api.UserFeignClient;
import com.njcn.user.pojo.po.User;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/**
* @author xy
*/
@Service
@Slf4j
@RequiredArgsConstructor
public class SmsNotificationService {
private final AppMsgSetFeignClient appMsgSetFeignClient;
private final UserFeignClient userFeignClient;
private final SmsSendFeignClient smsSendFeignClient;
@Value("${msg.msg_sign:南京灿能电力}")
private String msgSign;
@Async("smsNotificationExecutor")
public void asyncSendSmsNotification(String deviceId
, DevDetailDTO devDetailDto
, LocalDateTime eventTime
, Double amplitude
, Double persistTime) {
List<String> userIdList = appMsgSetFeignClient.queryUserIdsByDeviceId(deviceId).getData();
if (CollectionUtil.isNotEmpty(userIdList)) {
List<User> userList = userFeignClient.getUserListByIds(userIdList).getData();
if (CollectionUtil.isNotEmpty(userList)) {
List<User> userList1 = userList.stream()
.filter(item -> StrUtil.isNotBlank(item.getPhone()) && Objects.equals(item.getSmsNotice(), 1))
.collect(Collectors.toList());
if (CollectionUtil.isNotEmpty(userList1)) {
String msgContent = "【" + msgSign + "】" + devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+ "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂降事件"
+ "，特征幅值：" + amplitude + "%"
+ "，持续时间：" + persistTime + "s";
userList1.forEach(item -> {
smsSendFeignClient.sendSmsSimple(item.getPhone(), msgContent, "verify_code");
});
}
}
}
}
}

68
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/CsEventPOServiceImpl.java
View File

@@ -13,10 +13,11 @@ import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.advance.api.EventCauseFeignClient;
import com.njcn.advance.pojo.dto.EventAnalysisDTO;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.CsLedgerFeignClient;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.DeviceMessageFeignClient;
import com.njcn.csdevice.enums.AlgorithmResponseEnum;
import com.njcn.csdevice.pojo.dto.DevDetailDTO;
import com.njcn.csdevice.pojo.po.CsLinePO;
@@ -33,11 +34,12 @@ import com.njcn.csharmonic.pojo.po.CsEventPO;
import com.njcn.csharmonic.pojo.vo.CsEventVO;
import com.njcn.csharmonic.pojo.vo.CsWarnDescVO;
import com.njcn.csharmonic.pojo.vo.EventDetailVO;
import com.njcn.csharmonic.service.AppNotificationService;
import com.njcn.csharmonic.service.CsEventPOService;
import com.njcn.csharmonic.service.CsEventUserPOService;
import com.njcn.csharmonic.service.SmsNotificationService;
import com.njcn.cssystem.api.MsgSendFeignClient;
import com.njcn.cssystem.pojo.param.MsgSendParam;
import com.njcn.event.common.mapper.WlRmpEventDetailMapper;
import com.njcn.event.common.service.EventAnalysisService;
import com.njcn.event.file.component.WaveFileComponent;
import com.njcn.event.file.component.WavePicComponent;
import com.njcn.event.file.pojo.bo.WaveDataDetail;
@@ -57,6 +59,7 @@ import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.api.EleEvtFeignClient;
import com.njcn.system.api.EpdFeignClient;
import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.enums.DicDataTypeEnum;
import com.njcn.system.pojo.po.DictData;
import com.njcn.system.pojo.po.EleEpdPqd;
import com.njcn.system.pojo.po.EleEvtParm;
@@ -71,7 +74,6 @@ import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
@@ -120,13 +122,9 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
private final DicDataFeignClient dicDataFeignClient;
private final WlRmpEventDetailMapper wlRmpEventDetailMapper;
private final CsLedgerFeignClient csLedgerFeignclient;
private final DeviceMessageFeignClient deviceMessageClient;
private final AppNotificationService appNotificationService;
private final SmsNotificationService smsNotificationService;
@Value("${msg.msg_sign:南京灿能电力}")
private String msgSign;
private final EventAnalysisService eventAnalysisService;
private final EventCauseFeignClient eventCauseFeignClient;
private final MsgSendFeignClient msgSendFeignClient;
@Override
public List<EventDetailVO> queryEventList(CsEventUserQueryParam csEventUserQueryParam) {
@@ -393,6 +391,8 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
eventPo.setPersistTime(param.getDuration());
eventPo.setAmplitude(param.getAmplitude() * 100);
eventPo.setPhase(param.getPhase());
String dropZone = eventAnalysisService.determineDropZone(String.valueOf(param.getAmplitude() * 100),String.valueOf(param.getDuration()));
eventPo.setLandPoint(dropZone);
this.baseMapper.insert(eventPo);
//influxDB数据录入
List<String> records = new ArrayList<String>();
@@ -412,8 +412,6 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
//同步数据到 r_mp_event_detail
insertEvent(uuid, param, time);
//根据设备获取主用户、子用户
List<String> eventUser = deviceMessageClient.getEventUserByDeviceId(po.getDeviceId(),true).getData();
//获取台账信息
DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(po.getDeviceId()).getData();
LocalDateTime eventTime = LocalDateTime.parse(param.getStartTime(), DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN));
@@ -423,7 +421,18 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
//异步处理事件用户关系和App消息推送
try {
appNotificationService.asyncSaveEventUserAndNotify(uuid, eventUser, devDetailDto, time, param.getEventType(), amplitudePercent, durationSeconds);
MsgSendParam msgSendParam = new MsgSendParam();
msgSendParam.setEventType(1);
msgSendParam.setType("2");
msgSendParam.setDevId(po.getDeviceId());
msgSendParam.setEventName(getTag(param.getEventType()));
msgSendParam.setEventTime(eventTime);
msgSendParam.setId(uuid);
msgSendParam.setNDid(devDetailDto.getNDid());
msgSendParam.setAmplitude(amplitudePercent);
msgSendParam.setPersistTime(durationSeconds);
msgSendParam.setDropZone(dropZone);
msgSendFeignClient.appMsgSend(msgSendParam);
} catch (Exception e) {
log.error("异步保存事件用户和通知失败，事件ID: {}", uuid, e);
}
@@ -431,22 +440,45 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
//如果是暂降事件，异步发送短信通知
if (param.getEventType() == 1) {
try {
smsNotificationService.asyncSendSmsNotification(po.getDeviceId(), devDetailDto, eventTime, amplitudePercent, durationSeconds);
MsgSendParam msgSendParam = new MsgSendParam();
msgSendParam.setDevId(po.getDeviceId());
msgSendParam.setEventTime(eventTime);
msgSendParam.setAmplitude(amplitudePercent);
msgSendParam.setPersistTime(durationSeconds);
msgSendParam.setDropZone(dropZone);
msgSendFeignClient.smsMsgSend(msgSendParam);
} catch (Exception e) {
log.error("异步发送短信通知失败，事件ID: {}", uuid, e);
}
}
} else {
if (StrUtil.isNotBlank(param.getWavePath())) {
EventAnalysisDTO var1 = new EventAnalysisDTO();
var1.setWlFilePath(param.getWavePath());
EventAnalysisDTO dto = eventCauseFeignClient.analysisCauseAndType(var1).getData();
List<DictData> list1 = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.EVENT_REASON.getCode()).getData();
String id1 = list1.stream()
.filter(item -> Objects.equals(item.getAlgoDescribe(), dto.getCause()))
.map(DictData::getId)
.findFirst()
.orElse(null);
List<DictData> list2 = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.EVENT_TYPE.getCode()).getData();
String id2 = list2.stream()
.filter(item -> Objects.equals(item.getAlgoDescribe(), dto.getType()))
.map(DictData::getId)
.findFirst()
.orElse(null);
//更新文件信息
//先校验两份文件的名称是否一致
this.lambdaUpdate()
.eq(CsEventPO::getLineId,param.getMonitorId())
.eq(CsEventPO::getStartTime,param.getStartTime())
.set(CsEventPO::getWavePath,param.getWavePath())
.set(CsEventPO::getAdvanceReason,id1)
.set(CsEventPO::getAdvanceType,id2)
.update();
//更新文件信息
updateEvent(param);
updateEvent(param,id1,id2);
}
}
}
@@ -486,11 +518,13 @@ public class CsEventPOServiceImpl extends ServiceImpl<CsEventPOMapper, CsEventPO
wlRmpEventDetailMapper.insert(rmpEventDetailPO);
}
public void updateEvent(CldEventParam param) {
public void updateEvent(CldEventParam param,String advanceReason, String advanceType) {
LambdaQueryWrapper<RmpEventDetailPO> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(RmpEventDetailPO::getMeasurementPointId,param.getMonitorId()).eq(RmpEventDetailPO::getStartTime,param.getStartTime());
RmpEventDetailPO po = wlRmpEventDetailMapper.selectOne(wrapper);
po.setWavePath(param.getWavePath());
po.setAdvanceReason(advanceReason);
po.setAdvanceType(advanceType);
po.setFileFlag(1);
wlRmpEventDetailMapper.updateById(po);
}