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

refactor(event): 重构事件通知服务并优化异步处理

Browse Source 
- 新增 AppNotificationService 处理应用通知推送逻辑
- 新增 SmsNotificationService 处理短信通知异步发送
- 添加 AsyncConfig 配置类定义事件和短信通知线程池
- 在 CsAlarmServiceImpl 中替换 sendEventUtils 为 appNotificationService
- 在 EventServiceImpl 中替换 sendEventUtils 为 appNotificationService 和 smsNotificationService
- 移除废弃的 SendEventUtils 工具类
- 优化暂降事件短信发送逻辑并支持幅值和持续时间参数
- 修复时序数据库时间处理注释说明，统一使用UTC时间转换
- 启用 Spring 异步注解支持事件和短信异步处理
This commit is contained in:
xy
2026-04-22 16:05:01 +08:00
parent 02d321d4d4
commit 6dcee1f6c3
7 changed files with 177 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
View File

@@ -214,7 +214,7 @@ public class StatServiceImpl implements IStatService {
//这边特殊处理，如果数据为3.14159，则将数据置为null
fields.put(dataArrayList.get(i).getName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i));
fields.put(InfluxDBTableConstant.IS_ABNORMAL,item.getDataTag());
//fixme 这边前置传递的应该是UTC时间,但是前置说是传递的北京时间,讨论了一下没太理解。这边暂时先这样处理,influx入库处理成北京时间,减去8小时。
//fixme 设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时
boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), devType) && Objects.equals(accessMethod, "CLD");
Point point = influxDbUtils.pointBuilder(tableName, flag?item.getDataTimeSec():item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();

2
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/ZlEventBootApplication.java
View File

@@ -6,6 +6,7 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.DependsOn;
import org.springframework.scheduling.annotation.EnableAsync;
/**
@@ -18,6 +19,7 @@ import org.springframework.context.annotation.DependsOn;
@MapperScan("com.njcn.**.mapper")
@EnableFeignClients(basePackages = "com.njcn")
@SpringBootApplication(scanBasePackages = "com.njcn")
@EnableAsync
public class ZlEventBootApplication {
public static void main(String[] args) {

42
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/config/AsyncConfig.java Normal file
View File

@@ -0,0 +1,42 @@
package com.njcn.zlevent.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;
}
}

69
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/utils/SendEventUtils.java → iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/AppNotificationService.java
View File

@@ -1,4 +1,4 @@
package com.njcn.zlevent.utils;
package com.njcn.zlevent.service;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
@@ -14,12 +14,11 @@ import com.njcn.csdevice.pojo.po.CsEventSendMsg;
import com.njcn.csharmonic.pojo.po.CsEventUserPO;
import com.njcn.system.api.EpdFeignClient;
import com.njcn.user.pojo.po.User;
import com.njcn.zlevent.service.ICsEventUserService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
@@ -27,42 +26,25 @@ import java.util.Objects;
import java.util.stream.Collectors;
/**
* 类的介绍：
*
* @author xuyang
* @version 1.0.0
* @createTime 2023/9/25 16:08
* @author xy
*/
@Service
@Slf4j
@Component
public class SendEventUtils {
@RequiredArgsConstructor
public class AppNotificationService {
@Resource
private EventLogsFeignClient eventLogsFeignClient;
@Resource
private EpdFeignClient epdFeignClient;
@Resource
private ICsEventUserService csEventUserService;
@Resource
private CsLedgerFeignClient csLedgerFeignclient;
@Resource
private SendMessageUtil sendMessageUtil;
@Resource
private EquipmentFeignClient equipmentFeignClient;
@Resource
private DeviceMessageFeignClient deviceMessageFeignClient;
private final DeviceMessageFeignClient deviceMessageFeignClient;
private final EquipmentFeignClient equipmentFeignClient;
private final EpdFeignClient epdFeignClient;
private final ICsEventUserService csEventUserService;
private final SendMessageUtil sendMessageUtil;
private final EventLogsFeignClient eventLogsFeignClient;
private final CsLedgerFeignClient csLedgerFeignClient;
/**
* 事件推送给相关用户
* @param eventType 事件类型 1:事件 2:告警
* @param type 等级 事件分为设备事件、暂态事件、稳态事件 告警分为Ⅰ级告警、Ⅱ级告警、Ⅲ级告警
* @param devId 设备id
* @param eventName 事件名称
* @param eventTime 事件发生事件
* @param id 事件id
*/
@Transactional(rollbackFor = Exception.class)
public void sendUser(Integer eventType,String type,String devId, String eventName, LocalDateTime eventTime, String id, String nDid) {
@Async("eventNotificationExecutor")
public void sendAppNotification(Integer eventType, String type, String devId,
String eventName, LocalDateTime eventTime,
String id, String nDid, Double amplitude, Double persistTime) {
int code;
List<User> users = new ArrayList<>();
List<String> eventUser;
@@ -71,11 +53,12 @@ public class SendEventUtils {
List<CsEventSendMsg> csEventSendMsgList = new ArrayList<>();
NoticeUserDto noticeUserDto = new NoticeUserDto();
NoticeUserDto.Payload payload = new NoticeUserDto.Payload();
String content;
String content = null;
List<CsEventUserPO> result = new ArrayList<>();
//获取设备类型 true:治理设备 false:其他类型的设备
boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
if (devModel) {
DevDetailDTO devDetailDto = csLedgerFeignClient.queryDevDetail(devId).getData();
//事件处理
if (eventType == 1){
eventName = epdFeignClient.findByName(eventName).getData().getShowName();
@@ -126,6 +109,11 @@ public class SendEventUtils {
devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
noticeUserDto.setPushClientId(devCodeList);
noticeUserDto.setTitle("暂态事件");
content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+ "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂态事件，事件类型："
+ eventName
+ "，特征幅值：" + amplitude + "%"
+ "，持续时间：" + persistTime + "s";
}
}
break;
@@ -154,8 +142,9 @@ public class SendEventUtils {
break;
}
//获取台账信息
DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(devId).getData();
content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生" + eventName;
if (Objects.isNull(content)) {
content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生" + eventName;
}
noticeUserDto.setContent(content);
payload.setType(code);
payload.setPath("/pages/index/message1?type="+payload.getType());

70
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/SmsNotificationService.java Normal file
View File

@@ -0,0 +1,70 @@
package com.njcn.zlevent.service;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.StrUtil;
import com.njcn.csdevice.api.CsLedgerFeignClient;
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;
private final CsLedgerFeignClient csLedgerFeignclient;
@Value("${msg.msg_sign:南京灿能电力}")
private String msgSign;
@Async("smsNotificationExecutor")
public void sendSmsForDipEvent(String deviceId, LocalDateTime eventTime,double amplitude,double persistTime) {
try {
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)) {
DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(deviceId).getData();
String msgContent = "【" + msgSign + "】" + devDetailDto.getEngineeringName()
+ "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+ "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂降事件"
+ "，特征幅值：" + amplitude + "%"
+ "，持续时间：" + persistTime + "s";
userList1.forEach(item -> {
try {
smsSendFeignClient.sendSmsSimple(item.getPhone(), msgContent, "verify_code");
} catch (Exception e) {
log.error("发送短信失败，手机号: {}", item.getPhone(), e);
}
});
}
}
}
} catch (Exception e) {
log.error("异步发送暂降事件短信失败，设备ID: {}", deviceId, e);
}
}
}

10
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsAlarmServiceImpl.java
View File

@@ -3,9 +3,7 @@ package com.njcn.zlevent.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.access.pojo.po.CsDeviceOnlineLogs;
import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.EquipmentFeignClient;
@@ -21,10 +19,10 @@ import com.njcn.system.pojo.po.EleEpdPqd;
import com.njcn.system.pojo.po.SysDicTreePO;
import com.njcn.zlevent.mapper.CsEventMapper;
import com.njcn.zlevent.pojo.po.CsEventLogs;
import com.njcn.zlevent.service.AppNotificationService;
import com.njcn.zlevent.service.ICsAlarmService;
import com.njcn.zlevent.service.ICsEventLogsService;
import com.njcn.zlevent.service.ICsEventService;
import com.njcn.zlevent.utils.SendEventUtils;
import lombok.AllArgsConstructor;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
@@ -52,11 +50,11 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
private final EquipmentFeignClient equipmentFeignClient;
private final EventServiceImpl eventService;
private final ICsEventService csEventService;
private final SendEventUtils sendEventUtils;
private final ICsEventLogsService csEventLogsService;
private final EpdFeignClient epdFeignClient;
private final RedisUtil redisUtil;
private final ChannelObjectUtil channelObjectUtil;
private final AppNotificationService appNotificationService;
@Override
@Transactional(rollbackFor = Exception.class)
@@ -128,9 +126,9 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
//推送事件逻辑处理 && cs_event_user入库 && 修改字典中告警事件的编码
for (AppEventMessage.DataArray item : dataArray) {
if (Objects.isNull(item.getCode())){
sendEventUtils.sendUser(2,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid());
appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid(),null,null);
} else {
sendEventUtils.sendUser(2,item.getType(),po.getId(),item.getCode(),eventTime,id,po.getNdid());
appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getCode(),eventTime,id,po.getNdid(),null, null);
//更新字典信息
EleEpdPqd eleEpdPqd = epdFeignClient.findByName(item.getName()).getData();
EleEpdPqdParam.EleEpdPqdUpdateParam updateParam = new EleEpdPqdParam.EleEpdPqdUpdateParam();

70
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java
View File

@@ -1,23 +1,22 @@
package com.njcn.zlevent.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.*;
import com.njcn.csdevice.pojo.dto.DevDetailDTO;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.DeviceMessageFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient;
import com.njcn.csdevice.api.WlRecordFeignClient;
import com.njcn.csdevice.pojo.param.WlRecordParam;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.csdevice.pojo.po.WlRecord;
import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
import com.njcn.csharmonic.pojo.po.CsEventPO;
import com.njcn.cssystem.api.AppMsgSetFeignClient;
import com.njcn.event.common.mapper.WlRmpEventDetailMapper;
import com.njcn.event.pojo.po.RmpEventDetailPO;
import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
@@ -33,19 +32,17 @@ import com.njcn.system.api.EpdFeignClient;
import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.pojo.dto.EpdDTO;
import com.njcn.system.pojo.po.DictData;
import com.njcn.user.api.UserFeignClient;
import com.njcn.user.pojo.po.User;
import com.njcn.zlevent.pojo.constant.ZlConstant;
import com.njcn.zlevent.service.AppNotificationService;
import com.njcn.zlevent.service.ICsEventService;
import com.njcn.zlevent.service.IEventService;
import com.njcn.zlevent.utils.SendEventUtils;
import com.njcn.zlevent.service.SmsNotificationService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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;
@@ -78,17 +75,12 @@ public class EventServiceImpl implements IEventService {
private final ICsEventService csEventService;
private final EquipmentFeignClient equipmentFeignClient;
private final InfluxDbUtils influxDbUtils;
private final SendEventUtils sendEventUtils;
private final WlRecordFeignClient wlRecordFeignClient;
private final WlRmpEventDetailMapper wlRmpEventDetailMapper;
private final DictTreeFeignClient dictTreeFeignClient;
private final DeviceMessageFeignClient deviceMessageFeignClient;
private final AppMsgSetFeignClient appMsgSetFeignClient;
private final UserFeignClient userFeignClient;
private final SmsSendFeignClient smsSendFeignClient;
private final CsLedgerFeignClient csLedgerFeignclient;
@Value("${msg.msg_sign:南京灿能电力}")
private String msgSign;
private final AppNotificationService appNotificationService;
private final SmsNotificationService smsNotificationService;
@Override
@DSTransactional
@@ -194,7 +186,7 @@ public class EventServiceImpl implements IEventService {
csEvent.setLocation("load");
}
}
//fixme 这边前置传递的应该是UTC时间,但是前置说是传递的北京时间,讨论了一下没太理解。这边暂时先这样处理,influx入库处理成北京时间,减去8小时。
//fixme 设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时
Point point = influxDbUtils.pointBuilder(tableName, item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
batchPoints.point(point);
@@ -203,6 +195,10 @@ public class EventServiceImpl implements IEventService {
list1.add(csEvent);
}
}
//evt_data入库
if (CollectionUtil.isNotEmpty(records)) {
influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, records);
}
//cs_event入库
if (CollectionUtil.isNotEmpty(list1)){
csEventService.saveBatch(list1);
@@ -211,36 +207,28 @@ public class EventServiceImpl implements IEventService {
if (CollectionUtil.isNotEmpty(filterList)) {
filterList.forEach(this::insertEvent);
}
//推送事件逻辑处理 && cs_event_user入库
//异步推送事件逻辑处理 && cs_event_user入库
for (AppEventMessage.DataArray item : dataArray) {
sendEventUtils.sendUser(1,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid());
//如果是暂降事件，则发送短信
if (Objects.equals(item.getName(), "Evt_Sys_DipStr")) {
//根据设备获取需要推送的用户列表
List<String> userIdList = appMsgSetFeignClient.queryUserIdsByDeviceId(po.getId()).getData();
if (CollectionUtil.isNotEmpty(userIdList)) {
//获取用户详细信息
List<User> userList = userFeignClient.getUserListByIds(userIdList).getData();
if (CollectionUtil.isNotEmpty(userList)) {
//筛选出有手机号码的；打开短信推送的
List<User> userList1 = userList.stream().filter(item2-> StrUtil.isNotBlank(item2.getPhone()) && Objects.equals(item2.getSmsNotice(),1)).collect(Collectors.toList());
if (CollectionUtil.isNotEmpty(userList1)) {
DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(po.getId()).getData();
String msgContent = "【"+msgSign+"】" +devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+ "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂降事件";
userList1.forEach(item2->{
smsSendFeignClient.sendSmsSimple(item2.getPhone(),msgContent, "verify_code");
});
}
double amplitude = 0.0;
double persistTime = 0.0;
List<AppEventMessage.Param> params = item.getParam();
if (CollectionUtil.isNotEmpty(params)) {
for (AppEventMessage.Param param : params) {
if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_VVADEPTH)) {
amplitude = Double.parseDouble(String.format("%.2f", Double.parseDouble(param.getData().toString())));
}
if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)) {
persistTime = Double.parseDouble(String.format("%.2f", Double.parseDouble(param.getData().toString())));
}
}
}
appNotificationService.sendAppNotification(1, item.getType(), po.getId(), item.getName(), eventTime, id, po.getNdid(),amplitude,persistTime);
//如果是暂降事件，则异步发送短信
if (Objects.equals(item.getName(), "Evt_Sys_DipStr")) {
smsNotificationService.sendSmsForDipEvent(po.getId(), eventTime,amplitude,persistTime);
}
}
}
//evt_data入库
if (CollectionUtil.isNotEmpty(records)) {
influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, records);
}
} catch (Exception e) {
log.error("事件入库异常：{}",e.getMessage());
}