Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
271c178a0f | ||
|
|
66d5364317 | ||
|
|
828630e424 | ||
|
|
be6628fe35 | ||
|
|
6c7ca8eb46 | ||
|
|
5dff7c6ff8 | ||
| a96fce38be |
@@ -54,17 +54,31 @@ public class EventGateServiceImpl implements EventGateService {
|
|||||||
String lineid = eventDetail.getLineId();
|
String lineid = eventDetail.getLineId();
|
||||||
LedgerBaseInfo pqLine = ledgerScaleMapper.getLedgerBaseInfo(Stream.of(lineid).collect(Collectors.toList())).get(0);
|
LedgerBaseInfo pqLine = ledgerScaleMapper.getLedgerBaseInfo(Stream.of(lineid).collect(Collectors.toList())).get(0);
|
||||||
String waveName = eventDetail.getWavePath();
|
String waveName = eventDetail.getWavePath();
|
||||||
String cfgPath, datPath;
|
String cfgPath, datPath,cfgPath1,datPath1;
|
||||||
|
InputStream cfgStream,datStream;
|
||||||
if (StrUtil.isBlank(waveName)) {
|
if (StrUtil.isBlank(waveName)) {
|
||||||
throw new BusinessException(WaveFileResponseEnum.ANALYSE_WAVE_NOT_FOUND);
|
throw new BusinessException(WaveFileResponseEnum.ANALYSE_WAVE_NOT_FOUND);
|
||||||
}
|
}
|
||||||
|
try {
|
||||||
cfgPath = generalInfo.getBusinessWavePath()+ File.separator+pqLine.getIp()+"/"+waveName+".CFG";
|
cfgPath = generalInfo.getBusinessWavePath()+ File.separator+pqLine.getIp()+"/"+waveName+".CFG";
|
||||||
datPath = generalInfo.getBusinessWavePath()+ File.separator+pqLine.getIp()+"/"+waveName+".DAT";
|
datPath = generalInfo.getBusinessWavePath()+ File.separator+pqLine.getIp()+"/"+waveName+".DAT";
|
||||||
|
cfgStream = waveFileComponent.getFileInputStreamByFilePath(cfgPath);
|
||||||
|
datStream = waveFileComponent.getFileInputStreamByFilePath(datPath);
|
||||||
log.info("本地磁盘波形文件路径----" + cfgPath);
|
log.info("本地磁盘波形文件路径----" + cfgPath);
|
||||||
InputStream cfgStream = waveFileComponent.getFileInputStreamByFilePath(cfgPath);
|
|
||||||
InputStream datStream = waveFileComponent.getFileInputStreamByFilePath(datPath);
|
}catch (Exception e){
|
||||||
|
cfgPath1 = generalInfo.getBusinessWavePath()+ File.separator+pqLine.getIp()+"/"+waveName+".cfg";
|
||||||
|
datPath1 = generalInfo.getBusinessWavePath()+ File.separator+pqLine.getIp()+"/"+waveName+".dat";
|
||||||
|
cfgStream = waveFileComponent.getFileInputStreamByFilePath(cfgPath1);
|
||||||
|
datStream = waveFileComponent.getFileInputStreamByFilePath(datPath1);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
if (Objects.isNull(cfgStream) || Objects.isNull(datStream)) {
|
if (Objects.isNull(cfgStream) || Objects.isNull(datStream)) {
|
||||||
|
|
||||||
throw new BusinessException(WaveFileResponseEnum.ANALYSE_WAVE_NOT_FOUND);
|
throw new BusinessException(WaveFileResponseEnum.ANALYSE_WAVE_NOT_FOUND);
|
||||||
|
|
||||||
}
|
}
|
||||||
waveDataDTO = waveFileComponent.getComtrade(cfgStream, datStream, 1);
|
waveDataDTO = waveFileComponent.getComtrade(cfgStream, datStream, 1);
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package com.njcn.product.event.transientes.controller;
|
package com.njcn.product.event.transientes.controller;
|
||||||
|
|
||||||
import cn.hutool.core.collection.CollUtil;
|
import cn.hutool.core.collection.CollUtil;
|
||||||
|
import cn.hutool.core.collection.CollectionUtil;
|
||||||
import cn.hutool.core.date.DatePattern;
|
import cn.hutool.core.date.DatePattern;
|
||||||
import cn.hutool.core.date.DateUtil;
|
import cn.hutool.core.date.DateUtil;
|
||||||
import cn.hutool.core.date.LocalDateTimeUtil;
|
import cn.hutool.core.date.LocalDateTimeUtil;
|
||||||
@@ -71,6 +72,8 @@ public class EventGateController extends BaseController {
|
|||||||
@Value("${SYS_TYPE_ZT}")
|
@Value("${SYS_TYPE_ZT}")
|
||||||
private String sysTypeZt;
|
private String sysTypeZt;
|
||||||
|
|
||||||
|
@Value("${business.RealTimeSMSSwitch:false}")
|
||||||
|
private boolean realTimeSMSSwitch;
|
||||||
|
|
||||||
private final WebSocketServer webSocketServer;
|
private final WebSocketServer webSocketServer;
|
||||||
|
|
||||||
@@ -97,6 +100,17 @@ public class EventGateController extends BaseController {
|
|||||||
private final ThreadPoolTaskExecutor smsTaskExecutor;
|
private final ThreadPoolTaskExecutor smsTaskExecutor;
|
||||||
private final PqlineCache pqlineCache;
|
private final PqlineCache pqlineCache;
|
||||||
|
|
||||||
|
private final SendMessageService messageService;
|
||||||
|
|
||||||
|
@GetMapping("/testSendMessage")
|
||||||
|
@ApiOperation("接收远程推送的暂态事件")
|
||||||
|
public HttpResult<Object> SendMessage(@RequestParam("startTime") String startTime,@RequestParam("endtTime") String endtTime) {
|
||||||
|
String methodDescribe = getMethodDescribe("SendMessage");
|
||||||
|
messageService.sendMessage(LocalDateTimeUtil.parse(startTime,DatePattern.NORM_DATETIME_PATTERN),LocalDateTimeUtil.parse(endtTime,DatePattern.NORM_DATETIME_PATTERN));
|
||||||
|
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
@OperateInfo
|
@OperateInfo
|
||||||
@GetMapping("/eventMsg")
|
@GetMapping("/eventMsg")
|
||||||
@ApiOperation("接收远程推送的暂态事件")
|
@ApiOperation("接收远程推送的暂态事件")
|
||||||
@@ -116,16 +130,34 @@ public class EventGateController extends BaseController {
|
|||||||
&& Float.parseFloat(jsonObject.get("eventvalue").toString()) <= msgEventConfigService.getEventValue()
|
&& Float.parseFloat(jsonObject.get("eventvalue").toString()) <= msgEventConfigService.getEventValue()
|
||||||
&& (Float.parseFloat(jsonObject.get("persisttime").toString()) * 1000) >= msgEventConfigService.getEventDuration()) {
|
&& (Float.parseFloat(jsonObject.get("persisttime").toString()) * 1000) >= msgEventConfigService.getEventDuration()) {
|
||||||
//过滤重要暂降事件
|
//过滤重要暂降事件
|
||||||
|
//F47过滤
|
||||||
jsonObject.set("persisttime",new BigDecimal(jsonObject.get("persisttime").toString()).setScale(3,RoundingMode.HALF_UP).toString());
|
float eventvalue = Float.parseFloat(jsonObject.get("eventvalue").toString());
|
||||||
|
float persisttime = Float.parseFloat(jsonObject.get("persisttime").toString()) * 1000;
|
||||||
|
if(shouldSendSMS(eventvalue,persisttime)){
|
||||||
Integer lineId = Integer.valueOf(jsonObject.get("lineid").toString());
|
Integer lineId = Integer.valueOf(jsonObject.get("lineid").toString());
|
||||||
List<PqUserLineAssPO> assList = pqUserLineAssMapper.selectList(new LambdaQueryWrapper<PqUserLineAssPO>().eq(PqUserLineAssPO::getLineIndex, lineId));
|
List<PqUserLineAssPO> assList = pqUserLineAssMapper.selectList(new LambdaQueryWrapper<PqUserLineAssPO>().eq(PqUserLineAssPO::getLineIndex, lineId));
|
||||||
|
|
||||||
String str ;
|
String str ;
|
||||||
if (CollUtil.isNotEmpty(assList)) {
|
if (CollUtil.isNotEmpty(assList)) {
|
||||||
List<String> userIds = assList.stream().map(PqUserLineAssPO::getUserIndex).distinct().collect(Collectors.toList());
|
List<String> userIds = assList.stream().map(PqUserLineAssPO::getUserIndex).distinct().collect(Collectors.toList());
|
||||||
List<PqUserLedgerPO> poList = pqUserLedgerMapper.selectList(new LambdaQueryWrapper<PqUserLedgerPO>().select(PqUserLedgerPO::getId, PqUserLedgerPO::getCustomerName).in(PqUserLedgerPO::getId, userIds));
|
List<PqUserLedgerPO> poList = pqUserLedgerMapper.selectList(new LambdaQueryWrapper<PqUserLedgerPO>().select(PqUserLedgerPO::getId, PqUserLedgerPO::getCustomerName,PqUserLedgerPO::getIsShow).in(PqUserLedgerPO::getId, userIds));
|
||||||
str = poList.stream().map(PqUserLedgerPO::getCustomerName).collect(Collectors.joining(";"));
|
List<PqUserLedgerPO> isShowUser = poList.stream().filter(temp -> temp.getIsShow() == 1).collect(Collectors.toList());
|
||||||
|
List<PqUserLedgerPO> noShowUser = poList.stream().filter(temp -> temp.getIsShow() != 1).collect(Collectors.toList());
|
||||||
|
String strUser="" ;
|
||||||
|
|
||||||
|
if(CollectionUtil.isEmpty(isShowUser)){
|
||||||
|
strUser =strUser+"影响集成电路敏感用户:无;";
|
||||||
|
}else {
|
||||||
|
String sensitiveUser = isShowUser.stream().map(PqUserLedgerPO::getCustomerName).collect(Collectors.joining(";"));
|
||||||
|
strUser =strUser+"影响集成电路敏感用户:"+sensitiveUser+";";
|
||||||
|
}
|
||||||
|
if(CollectionUtil.isEmpty(noShowUser)){
|
||||||
|
strUser =strUser+"影响其他用户:无;";
|
||||||
|
}else {
|
||||||
|
String sensitiveUser = noShowUser.stream().map(PqUserLedgerPO::getCustomerName).collect(Collectors.joining(";"));
|
||||||
|
strUser =strUser+"影响其他用户:"+sensitiveUser+";";
|
||||||
|
}
|
||||||
|
str = strUser;
|
||||||
} else {
|
} else {
|
||||||
str = "/";
|
str = "/";
|
||||||
}
|
}
|
||||||
@@ -142,11 +174,18 @@ public class EventGateController extends BaseController {
|
|||||||
webSocketServer.sendMessageToAll(jsonObject.toString());
|
webSocketServer.sendMessageToAll(jsonObject.toString());
|
||||||
|
|
||||||
//针对前置推送的暂降事件进行短信发送功能
|
//针对前置推送的暂降事件进行短信发送功能
|
||||||
|
//开启实时短信功能默认关闭走批量
|
||||||
|
if(realTimeSMSSwitch){
|
||||||
smsTaskExecutor.execute(() -> {
|
smsTaskExecutor.execute(() -> {
|
||||||
sendMessage(jsonObject, str);
|
sendMessage(jsonObject, str);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
e.printStackTrace();
|
e.printStackTrace();
|
||||||
log.error("暂降json格式异常!{}", e.getMessage());
|
log.error("暂降json格式异常!{}", e.getMessage());
|
||||||
@@ -154,6 +193,27 @@ public class EventGateController extends BaseController {
|
|||||||
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
|
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private boolean shouldSendSMS( Float value , Float time ) {
|
||||||
|
|
||||||
|
|
||||||
|
// 条件1: 电压降至50%以下，持续时间超过20ms
|
||||||
|
if (value < 0.5 && time >= 20) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 条件2: 电压降至50%—70%，持续时间超过200ms
|
||||||
|
if (value >= 50 && value < 70 && time >= 200) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 条件3: 电压降至70%—80%，持续时间超过500ms
|
||||||
|
if (value >= 70 && value < 80 && time >= 500) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
public Set<String> getAllParentIdsRecursive(String deptId, Map<String, PqsDepts> deptMap, Set<String> result) {
|
public Set<String> getAllParentIdsRecursive(String deptId, Map<String, PqsDepts> deptMap, Set<String> result) {
|
||||||
if (deptId == null || result.contains(deptId)) {
|
if (deptId == null || result.contains(deptId)) {
|
||||||
return result;
|
return result;
|
||||||
@@ -187,8 +247,12 @@ public class EventGateController extends BaseController {
|
|||||||
try {
|
try {
|
||||||
//下面一行代码正式环境需要放开
|
//下面一行代码正式环境需要放开
|
||||||
jsonObject = test();
|
jsonObject = test();
|
||||||
|
List<PqsDepts> deptsList = (List<PqsDepts>) redisUtil.getObjectByKey(REDIS_DEPT_INDEX + StrUtil.DASHED + "AllDept");
|
||||||
|
List<String> deptIds = deptsList.stream().map(PqsDepts::getDeptsIndex).distinct().collect(Collectors.toList());
|
||||||
|
jsonObject.putOpt("dept", String.join(StrUtil.COMMA,deptIds));
|
||||||
|
|
||||||
if (msgEventConfigService.getEventType().contains(jsonObject.get("wavetype").toString()) && Float.parseFloat(jsonObject.get("eventvalue").toString()) <= msgEventConfigService.getEventValue()) {
|
if (msgEventConfigService.getEventType().contains(jsonObject.get("wavetype").toString()) && Float.parseFloat(jsonObject.get("eventvalue").toString()) <= msgEventConfigService.getEventValue()) {
|
||||||
|
log.info("122345678788999999----------测试发送数据成功: {}",jsonObject);
|
||||||
webSocketServer.sendMessageToAll(jsonObject.toString());
|
webSocketServer.sendMessageToAll(jsonObject.toString());
|
||||||
|
|
||||||
//开始发送短信
|
//开始发送短信
|
||||||
@@ -232,8 +296,19 @@ public class EventGateController extends BaseController {
|
|||||||
//以下部分为测试数据后续删除
|
//以下部分为测试数据后续删除
|
||||||
List<PqLine> pqLineList = pqLineMapper.selectList(new LambdaQueryWrapper<>());
|
List<PqLine> pqLineList = pqLineMapper.selectList(new LambdaQueryWrapper<>());
|
||||||
List<Integer> lineList = pqLineList.stream().map(PqLine::getLineIndex).collect(Collectors.toList());
|
List<Integer> lineList = pqLineList.stream().map(PqLine::getLineIndex).collect(Collectors.toList());
|
||||||
List<LedgerBaseInfoDTO> baseInfoDTOList = pqLineMapper.getBaseLineInfo(lineList);
|
|
||||||
Map<Integer, LedgerBaseInfoDTO> map = baseInfoDTOList.stream().collect(Collectors.toMap(LedgerBaseInfoDTO::getLineId, Function.identity()));
|
List<LedgerBaseInfoDTO> ledgerBaseInfoDTOS = new ArrayList<>();
|
||||||
|
if(lineList.size()>1000){
|
||||||
|
List<List<Integer>> listIds = CollUtil.split(lineList,1000);
|
||||||
|
for(List<Integer> itemIds : listIds){
|
||||||
|
List<LedgerBaseInfoDTO> temp =pqLineMapper.getBaseLineInfo(itemIds);
|
||||||
|
ledgerBaseInfoDTOS.addAll(temp);
|
||||||
|
}
|
||||||
|
}else {
|
||||||
|
List<LedgerBaseInfoDTO> temp =pqLineMapper.getBaseLineInfo(lineList);
|
||||||
|
ledgerBaseInfoDTOS.addAll(temp);
|
||||||
|
}
|
||||||
|
Map<Integer, LedgerBaseInfoDTO> map = ledgerBaseInfoDTOS.stream().collect(Collectors.toMap(LedgerBaseInfoDTO::getLineId, Function.identity()));
|
||||||
|
|
||||||
Random random = new Random();
|
Random random = new Random();
|
||||||
Integer lineId = lineList.get(random.nextInt(lineList.size()));
|
Integer lineId = lineList.get(random.nextInt(lineList.size()));
|
||||||
@@ -355,15 +430,13 @@ public class EventGateController extends BaseController {
|
|||||||
|
|
||||||
if (CollUtil.isNotEmpty(poList)) {
|
if (CollUtil.isNotEmpty(poList)) {
|
||||||
StringBuilder stringBuilder = new StringBuilder(jsonObject.get("timeid").toString());
|
StringBuilder stringBuilder = new StringBuilder(jsonObject.get("timeid").toString());
|
||||||
|
String busName = jsonObject.containsKey("busname")? "_"+jsonObject.get("busname").toString():"";
|
||||||
BigDecimal bigDecimal = new BigDecimal(jsonObject.get("eventvalue").toString()).multiply(new BigDecimal(100)).setScale(2, RoundingMode.HALF_UP);
|
BigDecimal bigDecimal = new BigDecimal(jsonObject.get("eventvalue").toString()).multiply(new BigDecimal(100)).setScale(2, RoundingMode.HALF_UP);
|
||||||
stringBuilder.append(".").append(jsonObject.get("ms").toString()).append(",").append(jsonObject.get("bdname").toString()).append("_").append(jsonObject.get("busname").toString()).append("_").append(jsonObject.get("pointname").toString())
|
stringBuilder.append(".").append(jsonObject.get("ms").toString()).append(",").append(jsonObject.get("bdname").toString()).append(busName)
|
||||||
.append("发生电压暂降事件,事件残余电压").append(bigDecimal).append("%,持续时间:").append(jsonObject.get("persisttime").toString()).append("S;影响用户:");
|
// .append("_").append(jsonObject.get("pointname").toString())
|
||||||
|
.append("发生电压暂降事件,事件残余电压").append(bigDecimal).append("%,持续时间:").append(String.format("%.3f", Double.parseDouble(jsonObject.get("persisttime").toString()))).append("S;");
|
||||||
stringBuilder.append(objStr);
|
stringBuilder.append(objStr);
|
||||||
|
|
||||||
//TODO 临时处理，后面需要用上
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
String message;
|
String message;
|
||||||
if (stringBuilder.length() > 500) {
|
if (stringBuilder.length() > 500) {
|
||||||
message = stringBuilder.substring(0, 490).concat(";详情请登录电压暂降监测平台查看。");
|
message = stringBuilder.substring(0, 490).concat(";详情请登录电压暂降监测平台查看。");
|
||||||
|
|||||||
@@ -0,0 +1,35 @@
|
|||||||
|
package com.njcn.product.event.transientes.job;
|
||||||
|
|
||||||
|
import com.njcn.product.event.transientes.service.SendMessageService;
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||||
|
import org.springframework.scheduling.annotation.Scheduled;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2026/05/29 上午 10:45【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
@EnableScheduling
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
@Slf4j
|
||||||
|
public class SendMessageJob {
|
||||||
|
private final SendMessageService messageService;
|
||||||
|
|
||||||
|
@Scheduled(cron = "${business.sendMessageCronExpression}")
|
||||||
|
public void executeJob() {
|
||||||
|
// 获取当前时间
|
||||||
|
LocalDateTime now = LocalDateTime.now();
|
||||||
|
// 减去一个小时
|
||||||
|
LocalDateTime threeHourAgo = now.minusHours(3);
|
||||||
|
|
||||||
|
messageService.sendMessage(threeHourAgo,now);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -34,7 +34,7 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter {
|
|||||||
protected void configure(HttpSecurity http) throws Exception {
|
protected void configure(HttpSecurity http) throws Exception {
|
||||||
http.csrf().disable()
|
http.csrf().disable()
|
||||||
.authorizeRequests()
|
.authorizeRequests()
|
||||||
.antMatchers("/cn_authenticate","/ws/**","/accept/testEvent","/accept/eventMsg","/accept/simpleTest","/accept/phoneSend","/accept/refreshIdCache","/largescreen/sycnUser").permitAll() // 允许访问认证接口
|
.antMatchers("/cn_authenticate","/ws/**","/accept/testEvent","/accept/eventMsg","/accept/simpleTest","/accept/phoneSend","/accept/refreshIdCache","/accept/testSendMessage","/largescreen/sycnUser").permitAll() // 允许访问认证接口
|
||||||
// .antMatchers("/**").permitAll() // 允许访问认证接口
|
// .antMatchers("/**").permitAll() // 允许访问认证接口
|
||||||
.anyRequest().authenticated()
|
.anyRequest().authenticated()
|
||||||
.and()
|
.and()
|
||||||
|
|||||||
@@ -0,0 +1,14 @@
|
|||||||
|
package com.njcn.product.event.transientes.service;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2026/05/29 上午 11:30【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
public interface SendMessageService {
|
||||||
|
void sendMessage(LocalDateTime threeHourAgo, LocalDateTime now);
|
||||||
|
}
|
||||||
@@ -697,13 +697,14 @@ public class LargeScreenCountServiceImpl implements LargeScreenCountService {
|
|||||||
DateTime start = DateUtil.beginOfDay(DateUtil.parse(largeScreenCountParam.getSearchBeginTime()));
|
DateTime start = DateUtil.beginOfDay(DateUtil.parse(largeScreenCountParam.getSearchBeginTime()));
|
||||||
DateTime end = DateUtil.endOfDay(DateUtil.parse(largeScreenCountParam.getSearchEndTime()));
|
DateTime end = DateUtil.endOfDay(DateUtil.parse(largeScreenCountParam.getSearchEndTime()));
|
||||||
|
|
||||||
LambdaQueryWrapper<MsgEventInfo> lambdaQueryWrapper = new LambdaQueryWrapper<>();
|
QueryWrapper<MsgEventInfo> queryWrapper = new QueryWrapper<>();
|
||||||
lambdaQueryWrapper.eq(!StringUtils.isEmpty(largeScreenCountParam.getSendResult()), MsgEventInfo::getSendResult, largeScreenCountParam.getSendResult());
|
queryWrapper.select("DISTINCT msg_index,user_id, user_name,send_time,msg_content,phone,send_result,is_handle ");
|
||||||
lambdaQueryWrapper.orderByDesc(MsgEventInfo::getSendTime).between(MsgEventInfo::getSendTime, start, end);
|
queryWrapper.lambda().eq(!StringUtils.isEmpty(largeScreenCountParam.getSendResult()), MsgEventInfo::getSendResult, largeScreenCountParam.getSendResult());
|
||||||
|
queryWrapper.lambda().orderByDesc(MsgEventInfo::getSendTime).between(MsgEventInfo::getSendTime, start, end);
|
||||||
if(StrUtil.isNotBlank(largeScreenCountParam.getSearchValue())){
|
if(StrUtil.isNotBlank(largeScreenCountParam.getSearchValue())){
|
||||||
lambdaQueryWrapper.and(w->w.like(MsgEventInfo::getUserName,largeScreenCountParam.getSearchValue()).or().like(MsgEventInfo::getPhone,largeScreenCountParam.getSearchValue()));
|
queryWrapper.lambda().and(w->w.like(MsgEventInfo::getUserName,largeScreenCountParam.getSearchValue()).or().like(MsgEventInfo::getPhone,largeScreenCountParam.getSearchValue()));
|
||||||
}
|
}
|
||||||
return msgEventInfoService.page(new Page<>(PageFactory.getPageNum(largeScreenCountParam), PageFactory.getPageSize(largeScreenCountParam)), lambdaQueryWrapper);
|
return msgEventInfoService.page(new Page<>(PageFactory.getPageNum(largeScreenCountParam), PageFactory.getPageSize(largeScreenCountParam)), queryWrapper);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|||||||
@@ -0,0 +1,409 @@
|
|||||||
|
package com.njcn.product.event.transientes.service.impl;
|
||||||
|
|
||||||
|
import cn.hutool.core.collection.CollUtil;
|
||||||
|
import cn.hutool.core.date.DatePattern;
|
||||||
|
import cn.hutool.core.date.LocalDateTimeUtil;
|
||||||
|
import cn.hutool.core.util.IdUtil;
|
||||||
|
import cn.hutool.core.util.StrUtil;
|
||||||
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||||
|
import com.njcn.product.event.config.RedisUtil;
|
||||||
|
import com.njcn.product.event.devcie.pojo.dto.LedgerBaseInfoDTO;
|
||||||
|
import com.njcn.product.event.devcie.pojo.po.PqsDeptsline;
|
||||||
|
import com.njcn.product.event.devcie.service.PqsDeptslineService;
|
||||||
|
import com.njcn.product.event.transientes.mapper.PqUserLedgerMapper;
|
||||||
|
import com.njcn.product.event.transientes.mapper.PqUserLineAssMapper;
|
||||||
|
import com.njcn.product.event.transientes.mapper.PqsEventdetailMapper;
|
||||||
|
import com.njcn.product.event.transientes.pojo.dto.SmsResponseDTO;
|
||||||
|
import com.njcn.product.event.transientes.pojo.dto.SmsSendDTO;
|
||||||
|
import com.njcn.product.event.transientes.pojo.po.*;
|
||||||
|
import com.njcn.product.event.transientes.service.*;
|
||||||
|
import com.njcn.product.event.transientes.utils.SmsUtils;
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
import org.springframework.util.CollectionUtils;
|
||||||
|
|
||||||
|
import java.math.BigDecimal;
|
||||||
|
import java.math.RoundingMode;
|
||||||
|
import java.time.Duration;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.*;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
import java.util.function.Function;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2026/05/29 上午 11:30【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Service
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
@Slf4j
|
||||||
|
public class SendMessageServiceImpl implements SendMessageService {
|
||||||
|
private final PqUserLineAssMapper pqUserLineAssMapper;
|
||||||
|
private final PqUserLedgerMapper pqUserLedgerMapper;
|
||||||
|
private final PqsEventdetailMapper pqsEventdetailMapper;
|
||||||
|
private final MsgEventInfoService msgEventInfoService;
|
||||||
|
private final static String NAME_KEY = "LineCache:";
|
||||||
|
private final RedisUtil redisUtil;
|
||||||
|
private final ThreadPoolTaskExecutor smsTaskExecutor;
|
||||||
|
private final PqsUsersetService pqsUsersetService;
|
||||||
|
private final SmsUtils smsUtils;
|
||||||
|
|
||||||
|
private final PqsDeptslineService pqsDeptslineService;
|
||||||
|
@Value("${SYS_TYPE_ZT}")
|
||||||
|
private String sysTypeZt;
|
||||||
|
|
||||||
|
private final PqsDeptsService pqsDeptsService;
|
||||||
|
|
||||||
|
private final PqsUserService pqsUserService;
|
||||||
|
@Override
|
||||||
|
public void sendMessage(LocalDateTime threeHourAgo, LocalDateTime now) {
|
||||||
|
log.info("开始扫描暂态事件时间段："+threeHourAgo+"-"+now);
|
||||||
|
//查询重要敏感客户
|
||||||
|
List<PqUserLedgerPO> poList = pqUserLedgerMapper.selectList(new LambdaQueryWrapper<PqUserLedgerPO>().select(PqUserLedgerPO::getId, PqUserLedgerPO::getCustomerName,PqUserLedgerPO::getIsShow).eq(PqUserLedgerPO::getIsShow, 1));
|
||||||
|
List<String> userIds = poList.stream().map(PqUserLedgerPO::getId).collect(Collectors.toList());
|
||||||
|
List<PqUserLineAssPO> pqUserLineAssPOS = pqUserLineAssMapper.selectList(new LambdaQueryWrapper<PqUserLineAssPO>().in(PqUserLineAssPO::getUserIndex, userIds));
|
||||||
|
List<Integer> lineIds = pqUserLineAssPOS.stream().map(PqUserLineAssPO::getLineIndex).distinct().collect(Collectors.toList());
|
||||||
|
LambdaQueryWrapper<PqsEventdetail> lambdaQueryWrapper = new LambdaQueryWrapper<>();
|
||||||
|
lambdaQueryWrapper.in(PqsEventdetail::getLineid,lineIds).between(PqsEventdetail::getCreateTime,threeHourAgo,now);
|
||||||
|
List<PqsEventdetail> pqsEventdetails = pqsEventdetailMapper.selectList(lambdaQueryWrapper);
|
||||||
|
log.info("扫描到敏感客户暂态事件："+pqsEventdetails.size()+"条");
|
||||||
|
|
||||||
|
if(!CollectionUtils.isEmpty(pqsEventdetails)){
|
||||||
|
List<LedgerBaseInfoDTO> ledgerBaseInfoDTOS = (List<LedgerBaseInfoDTO>)redisUtil.getObjectByKey(NAME_KEY + StrUtil.DASHED+"LedgerBaseInfoDTO");
|
||||||
|
Map<Integer, LedgerBaseInfoDTO> LedgerBaseInfoDTOMap = ledgerBaseInfoDTOS.stream().collect(Collectors.toMap(LedgerBaseInfoDTO::getLineId, Function.identity()));
|
||||||
|
|
||||||
|
List<MsgEventInfo> msgEventInfos = msgEventInfoService.lambdaQuery().between(MsgEventInfo::getSendTime, threeHourAgo, now).eq(MsgEventInfo::getSendResult, 1).list();
|
||||||
|
List<String> successSendEventIds = msgEventInfos.stream().map(MsgEventInfo::getEventIndex).distinct().collect(Collectors.toList());
|
||||||
|
|
||||||
|
pqsEventdetails = pqsEventdetails.stream()
|
||||||
|
.filter(temp -> shouldSendSMS(temp.getEventvalue(), temp.getPersisttime()) && (!successSendEventIds.contains(temp.getEventdetailIndex())))
|
||||||
|
.collect(Collectors.toList());
|
||||||
|
log.info("扫描到敏感客户暂态事件过滤后事件："+pqsEventdetails.size()+"条");
|
||||||
|
|
||||||
|
if(!CollectionUtils.isEmpty(pqsEventdetails)){
|
||||||
|
log.info("过滤后事件数："+pqsEventdetails.size());
|
||||||
|
|
||||||
|
List<PqsEventdetail> finalPqsEventdetails = pqsEventdetails;
|
||||||
|
poList.forEach(temp->{
|
||||||
|
List<Integer> tempLineIds = pqUserLineAssPOS.stream()
|
||||||
|
.filter(pqUserLineAssPO -> Objects.equals(pqUserLineAssPO.getUserIndex(), temp.getId()))
|
||||||
|
.map(PqUserLineAssPO::getLineIndex).collect(Collectors.toList());
|
||||||
|
if(CollectionUtils.isEmpty(tempLineIds)){
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
List<PqsEventdetail> result = finalPqsEventdetails.stream().filter(pqsEventdetail -> tempLineIds.contains(pqsEventdetail.getLineid())).collect(Collectors.toList());
|
||||||
|
if(CollectionUtils.isEmpty(result)){
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
//组装台账信息
|
||||||
|
List<LedgerBaseInfoDTO> tempLedger = ledgerBaseInfoDTOS.stream().filter(line -> tempLineIds.contains(line.getLineId())).collect(Collectors.toList());
|
||||||
|
|
||||||
|
Map<String, List<String>> busNameMap = tempLedger.stream().map(LedgerBaseInfoDTO::getBusBarName).collect(Collectors.groupingBy(SendMessageServiceImpl::extractVoltage));
|
||||||
|
String allBus = busNameMap.entrySet().stream().sorted(Map.Entry.comparingByKey()).map(
|
||||||
|
entry -> {
|
||||||
|
return entry.getKey() + "母线" + entry.getValue().size() + "条";
|
||||||
|
}
|
||||||
|
).collect(Collectors.joining("、"));
|
||||||
|
Map<String, List<String>> subStaionMap = tempLedger.stream().map(LedgerBaseInfoDTO::getStationName).distinct().collect(Collectors.groupingBy(SendMessageServiceImpl::extractVoltage));
|
||||||
|
|
||||||
|
String allsubStation = subStaionMap.entrySet().stream()
|
||||||
|
.sorted(Map.Entry.comparingByKey()) // 按电压等级排序（110,220,500）
|
||||||
|
.map(entry -> {
|
||||||
|
String voltage = entry.getKey(); // 例如 "110kV"
|
||||||
|
List<String> stations = entry.getValue();
|
||||||
|
// 构建组内字符串：第一个保留全名，后续的去掉电压前缀
|
||||||
|
StringBuilder sb = new StringBuilder();
|
||||||
|
for (int i = 0; i < stations.size(); i++) {
|
||||||
|
String fullName = stations.get(i);
|
||||||
|
if (i == 0) {
|
||||||
|
sb.append(fullName); // 第一个保留完整名称（含电压）
|
||||||
|
} else {
|
||||||
|
// 去掉电压前缀（例如 "110kV标厂变" -> "标厂变"）
|
||||||
|
String pureName = fullName.startsWith(voltage)
|
||||||
|
? fullName.substring(voltage.length())
|
||||||
|
: fullName;
|
||||||
|
sb.append("、").append(pureName);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return sb.toString();
|
||||||
|
})
|
||||||
|
.collect(Collectors.joining(","));
|
||||||
|
result.forEach(event->{
|
||||||
|
event.setBusBarName(LedgerBaseInfoDTOMap.get(event.getLineid()).getBusBarName());
|
||||||
|
event.setStationName(LedgerBaseInfoDTOMap.get(event.getLineid()).getStationName());
|
||||||
|
});
|
||||||
|
|
||||||
|
|
||||||
|
Map<String, List<PqsEventdetail>> groupFilterNull = result.stream()
|
||||||
|
.filter(e -> e.getStationName() != null)
|
||||||
|
.collect(Collectors.groupingBy(PqsEventdetail::getStationName));
|
||||||
|
//组装暂降信息
|
||||||
|
StringBuilder stringBuilder = new StringBuilder(temp.getCustomerName());
|
||||||
|
stringBuilder.append("上级直供变电站共有相关"+allBus+",");
|
||||||
|
stringBuilder.append("涉及"+allsubStation+"。");
|
||||||
|
AtomicInteger index = new AtomicInteger();
|
||||||
|
|
||||||
|
String eventString = groupFilterNull.entrySet().stream().map(entry -> {
|
||||||
|
String substationnName = entry.getKey(); // 例如 "110kV智芯变"
|
||||||
|
List<PqsEventdetail> value = entry.getValue();
|
||||||
|
StringBuilder sb = new StringBuilder();
|
||||||
|
if (index.getAndIncrement() == 0) {
|
||||||
|
sb.append(LocalDateTimeUtil.format(value.get(0).getTimeid(),DatePattern.NORM_DATETIME_MINUTE_PATTERN)).append(",");
|
||||||
|
}else {
|
||||||
|
sb.append(LocalDateTimeUtil.format(value.get(0).getTimeid(),"HH:mm")).append(",");
|
||||||
|
|
||||||
|
}
|
||||||
|
if (value.size() == 1) {
|
||||||
|
|
||||||
|
BigDecimal eventvalue = new BigDecimal(value.get(0).getEventvalue()).multiply(new BigDecimal(100)).setScale(2, RoundingMode.HALF_UP);
|
||||||
|
BigDecimal persisttime = new BigDecimal(value.get(0).getPersisttime()).divide(new BigDecimal(1000)).setScale(3, RoundingMode.HALF_UP);
|
||||||
|
|
||||||
|
sb.append(substationnName).append(value.get(0).getBusBarName()).append("发生电压暂降,电压跌落至").append(eventvalue).append("%,持续时间:").append(persisttime).append("S");
|
||||||
|
} else {
|
||||||
|
// String minTime = LocalDateTimeUtil.format(value.stream()
|
||||||
|
// .map(PqsEventdetail::getTimeid)
|
||||||
|
// .min(LocalDateTime::compareTo).get(), DatePattern.NORM_DATETIME_MINUTE_PATTERN);
|
||||||
|
// String maxTime = LocalDateTimeUtil.format(value.stream()
|
||||||
|
// .map(eventdetail -> {
|
||||||
|
// return eventdetail.getTimeid().plus(Duration.ofMillis(eventdetail.getMs().longValue()));
|
||||||
|
// })
|
||||||
|
// .max(LocalDateTime::compareTo).get(), DatePattern.NORM_DATETIME_MS_FORMATTER);
|
||||||
|
// String BusNameList = value.stream().map(PqsEventdetail::getBusBarName).collect(Collectors.joining("、"));
|
||||||
|
Map<String, List<String>> BusNameMap = value.stream().map(PqsEventdetail::getBusBarName).distinct().collect(Collectors.groupingBy(SendMessageServiceImpl::extractVoltage));
|
||||||
|
String allBusName = BusNameMap.entrySet().stream()
|
||||||
|
.sorted(Map.Entry.comparingByKey()) // 按电压等级排序（110,220,500）
|
||||||
|
.map(tempEntry -> {
|
||||||
|
String voltage = tempEntry.getKey(); // 例如 "110kV"
|
||||||
|
List<String> busNameList = tempEntry.getValue();
|
||||||
|
// 构建组内字符串：第一个保留全名，后续的去掉电压前缀
|
||||||
|
StringBuilder busName = new StringBuilder();
|
||||||
|
for (int i = 0; i < busNameList.size(); i++) {
|
||||||
|
String fullName = busNameList.get(i).replace("母线","");
|
||||||
|
if (i == 0) {
|
||||||
|
busName.append(fullName); // 第一个保留完整名称（含电压）
|
||||||
|
} else {
|
||||||
|
// 去掉电压前缀（例如 "10kV3B#母线" -> "3B#"）
|
||||||
|
String pureName = fullName.startsWith(voltage)
|
||||||
|
? fullName.substring(voltage.length())
|
||||||
|
: fullName;
|
||||||
|
busName.append("、").append(pureName.replace("母线",""));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return busName.toString();
|
||||||
|
})
|
||||||
|
.collect(Collectors.joining(","));
|
||||||
|
BigDecimal eventvalueMin = new BigDecimal(value.stream().mapToDouble(PqsEventdetail::getEventvalue).min().getAsDouble())
|
||||||
|
.multiply(new BigDecimal(100)).setScale(2, RoundingMode.HALF_UP);
|
||||||
|
BigDecimal eventvalueMax = new BigDecimal(value.stream().mapToDouble(PqsEventdetail::getEventvalue).max().getAsDouble())
|
||||||
|
.multiply(new BigDecimal(100)).setScale(2, RoundingMode.HALF_UP);
|
||||||
|
|
||||||
|
BigDecimal persisttimeMin = new BigDecimal(value.stream().mapToDouble(PqsEventdetail::getPersisttime).min().getAsDouble())
|
||||||
|
.divide(new BigDecimal(1000)).setScale(3, RoundingMode.HALF_UP);
|
||||||
|
BigDecimal persisttimeMax = new BigDecimal(value.stream().mapToDouble(PqsEventdetail::getPersisttime).max().getAsDouble())
|
||||||
|
.divide(new BigDecimal(1000)).setScale(3, RoundingMode.HALF_UP);
|
||||||
|
sb.append(substationnName).append(allBusName)
|
||||||
|
.append("母线发生电压暂降,电压跌落至").append(eventvalueMin).append("%—").append(eventvalueMax).append("%,持续时间:")
|
||||||
|
.append(persisttimeMin).append("S—").append(persisttimeMax).append("S");
|
||||||
|
}
|
||||||
|
|
||||||
|
return sb.toString();
|
||||||
|
}).collect(Collectors.joining(";"));
|
||||||
|
|
||||||
|
stringBuilder.append(eventString).append("。");
|
||||||
|
List<Integer> eventLineIds = result.stream().map(PqsEventdetail::getLineid).collect(Collectors.toList());
|
||||||
|
List<String> eventdetailIndexs = result.stream().map(PqsEventdetail::getEventdetailIndex).collect(Collectors.toList());
|
||||||
|
|
||||||
|
List<PqsDeptsline> pqLineDept = pqsDeptslineService.lambdaQuery().in(PqsDeptsline::getLineIndex, eventLineIds).eq(PqsDeptsline::getSystype, sysTypeZt).list();
|
||||||
|
if(CollectionUtils.isEmpty(pqLineDept)){
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Set<String> deptIds = pqLineDept.stream().map(PqsDeptsline::getDeptsIndex).collect(Collectors.toSet());
|
||||||
|
Set<String> resultIds = getAllParentDeptIds(deptIds);
|
||||||
|
|
||||||
|
List<PqsUserSet> pqsUserSetList = pqsUsersetService.lambdaQuery().eq(PqsUserSet::getIsNotice, 1).in(PqsUserSet::getDeptsIndex, resultIds).list();
|
||||||
|
if (CollUtil.isEmpty(pqsUserSetList)) {
|
||||||
|
//当前事件未找到用户信息，判断为不需要发送短信用户
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
List<PqsUser> pqsUserList = pqsUserService.lambdaQuery().select(PqsUser::getUserIndex, PqsUser::getPhone, PqsUser::getName).in(PqsUser::getUserIndex, pqsUserSetList.stream().map(PqsUserSet::getUserIndex).collect(Collectors.toList())).list();
|
||||||
|
if(!CollectionUtils.isEmpty(pqsUserList)){
|
||||||
|
//开始发送短信
|
||||||
|
smsTaskExecutor.execute(() -> {
|
||||||
|
sendMessageforUser(stringBuilder.toString(), pqsUserList,eventdetailIndexs);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
});
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
private void sendMessageforUser(String content, List<PqsUser> pqsUserList, List<String> eventdetailIndexs) {
|
||||||
|
|
||||||
|
//发送短信
|
||||||
|
List<MsgEventInfo> resultList = new ArrayList<>();
|
||||||
|
List<SmsSendDTO.ItemInner> msgDTOList = new ArrayList<>();
|
||||||
|
String message;
|
||||||
|
if (content.length() > 500) {
|
||||||
|
message = content.substring(0, 490).concat(";详情请登录电压暂降监测平台查看。");
|
||||||
|
} else {
|
||||||
|
message = content.toString();
|
||||||
|
}
|
||||||
|
for (PqsUser user : pqsUserList) {
|
||||||
|
String msgId = IdUtil.simpleUUID();
|
||||||
|
|
||||||
|
SmsSendDTO.ItemInner dto = new SmsSendDTO.ItemInner();
|
||||||
|
dto.setContent(message);
|
||||||
|
dto.setTo(user.getPhone());
|
||||||
|
dto.setCustomMsgID(msgId);
|
||||||
|
msgDTOList.add(dto);
|
||||||
|
for (String eventdetailIndex : eventdetailIndexs) {
|
||||||
|
MsgEventInfo msgEventInfo = new MsgEventInfo();
|
||||||
|
msgEventInfo.setMsgIndex(msgId);
|
||||||
|
msgEventInfo.setMsgContent(message);
|
||||||
|
msgEventInfo.setPhone(user.getPhone());
|
||||||
|
msgEventInfo.setUserId(user.getUserIndex());
|
||||||
|
msgEventInfo.setUserName(user.getName());
|
||||||
|
msgEventInfo.setIsHandle(0);
|
||||||
|
msgEventInfo.setSendResult(0);
|
||||||
|
msgEventInfo.setSendTime(LocalDateTime.now());
|
||||||
|
msgEventInfo.setEventIndex(eventdetailIndex);
|
||||||
|
resultList.add(msgEventInfo);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
List<SmsResponseDTO.SmsItem> result = smsUtils.sendSmSToUser(msgDTOList);
|
||||||
|
Map<String, SmsResponseDTO.SmsItem> stringSmsItemMap = result.stream().collect(Collectors.toMap(SmsResponseDTO.SmsItem::getCustomMsgID, Function.identity()));
|
||||||
|
|
||||||
|
resultList.forEach(item -> {
|
||||||
|
if (stringSmsItemMap.containsKey(item.getMsgIndex())) {
|
||||||
|
SmsResponseDTO.SmsItem smsItem = stringSmsItemMap.get(item.getMsgIndex());
|
||||||
|
item.setSendResult(Objects.equals(smsItem.getCode(), "0") ? 1 : 0);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
msgEventInfoService.saveBatch(resultList);
|
||||||
|
}
|
||||||
|
|
||||||
|
public Set<String> getAllParentDeptIds(Set<String> deptIds) {
|
||||||
|
// 首次获取直接父级
|
||||||
|
List<PqsDepts> allDeptList = pqsDeptsService.lambdaQuery().list();
|
||||||
|
// 递归获取所有父级
|
||||||
|
Set<String> result = recursivelyGetParentIds(deptIds, allDeptList);
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
/**
|
||||||
|
* 递归获取所有父级ID
|
||||||
|
*
|
||||||
|
* @param currentParentIds 当前层级的父级ID集合
|
||||||
|
* @return 所有层级的父级ID集合
|
||||||
|
*/
|
||||||
|
private Set<String> recursivelyGetParentIds(Set<String> currentParentIds, List<PqsDepts> allDeptList) {
|
||||||
|
Set<String> result = new HashSet<>(currentParentIds);
|
||||||
|
Set<String> nextLevelParentIds = new HashSet<>();
|
||||||
|
List<PqsDepts> parentDeptList = allDeptList.stream().filter(it -> currentParentIds.contains(it.getDeptsIndex())).collect(Collectors.toList());
|
||||||
|
for (PqsDepts pqsDepts : parentDeptList) {
|
||||||
|
if (!pqsDepts.getParentnodeid().equals("0")) {
|
||||||
|
nextLevelParentIds.add(pqsDepts.getParentnodeid());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// 如果有更高层级的父级，继续递归
|
||||||
|
if (!nextLevelParentIds.isEmpty()) {
|
||||||
|
Set<String> deeperParentIds = recursivelyGetParentIds(nextLevelParentIds, allDeptList);
|
||||||
|
result.addAll(deeperParentIds);
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
private boolean shouldSendSMS( Double value , Double time ) {
|
||||||
|
|
||||||
|
|
||||||
|
// 条件1: 电压降至50%以下，持续时间超过20ms
|
||||||
|
if (value < 0.5 && time >= 20) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 条件2: 电压降至50%—70%，持续时间超过200ms
|
||||||
|
if (value >= 50 && value < 70 && time >= 200) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 条件3: 电压降至70%—80%，持续时间超过500ms
|
||||||
|
if (value >= 70 && value < 80 && time >= 500) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String extractVoltage(String name) {
|
||||||
|
int kVIndex = name.indexOf("kV");
|
||||||
|
if (kVIndex == -1) {
|
||||||
|
log.info("存在台账"+name+"不符合命名规范");
|
||||||
|
throw new IllegalArgumentException("无法解析电压等级: " + name);
|
||||||
|
}
|
||||||
|
// 返回从开头到 "kV" 结束的部分（包含 "kV"）
|
||||||
|
return name.substring(0, kVIndex + 2);
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void main(String[] args) {
|
||||||
|
// Stream.of(1,2,3).collect(Collectors.toList()).forEach(
|
||||||
|
// temp->{
|
||||||
|
// if(temp==2){
|
||||||
|
// return;
|
||||||
|
// }
|
||||||
|
// System.out.println(temp);
|
||||||
|
// }
|
||||||
|
// );
|
||||||
|
List<String> substations = Arrays.asList(
|
||||||
|
"110kV智芯变",
|
||||||
|
"110kV标厂变",
|
||||||
|
"110kV科创变",
|
||||||
|
"220kV开发区变",
|
||||||
|
"220kV高新变",
|
||||||
|
"500kV枢纽换流站"
|
||||||
|
);
|
||||||
|
|
||||||
|
// 按电压等级字符串（含kV）分组
|
||||||
|
Map<String, List<String>> grouped = substations.stream()
|
||||||
|
.collect(Collectors.groupingBy(SendMessageServiceImpl::extractVoltage));
|
||||||
|
String collect = grouped.entrySet().stream()
|
||||||
|
.sorted(Map.Entry.comparingByKey()) // 按电压等级排序（110,220,500）
|
||||||
|
.map(entry -> {
|
||||||
|
String voltage = entry.getKey(); // 例如 "110kV"
|
||||||
|
List<String> stations = entry.getValue();
|
||||||
|
// 构建组内字符串：第一个保留全名，后续的去掉电压前缀
|
||||||
|
StringBuilder sb = new StringBuilder();
|
||||||
|
for (int i = 0; i < stations.size(); i++) {
|
||||||
|
String fullName = stations.get(i);
|
||||||
|
if (i == 0) {
|
||||||
|
sb.append(fullName); // 第一个保留完整名称（含电压）
|
||||||
|
} else {
|
||||||
|
// 去掉电压前缀（例如 "110kV标厂变" -> "标厂变"）
|
||||||
|
String pureName = fullName.startsWith(voltage)
|
||||||
|
? fullName.substring(voltage.length())
|
||||||
|
: fullName;
|
||||||
|
sb.append("、").append(pureName);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return sb.toString();
|
||||||
|
})
|
||||||
|
.collect(Collectors.joining(";"));
|
||||||
|
System.out.println(collect);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -40,11 +40,14 @@ business:
|
|||||||
wavePath: D://Comtrade
|
wavePath: D://Comtrade
|
||||||
targetPath: /pqmonitor
|
targetPath: /pqmonitor
|
||||||
exportBaseDir: D://exportComtrade
|
exportBaseDir: D://exportComtrade
|
||||||
eventCronExpression: 0 0/10 * * * ?
|
eventCronExpression: 0 0/2 * * * ?
|
||||||
failCronExpression: 0 5/10 * * * ?
|
failCronExpression: 0 5/10 * * * ?
|
||||||
userCronExpression: 0 5 1 * * ?
|
userCronExpression: 0 5 1 * * ?
|
||||||
syncinterval: 10
|
sendMessageCronExpression : 0 */3 * * * ?
|
||||||
|
syncinterval: 2
|
||||||
failsyncinterval: 1440
|
failsyncinterval: 1440
|
||||||
|
#实时短信功能
|
||||||
|
RealTimeSMSSwitch: false
|
||||||
#wavePath: /usr/local/comtrade
|
#wavePath: /usr/local/comtrade
|
||||||
#处理临时数据
|
#处理临时数据
|
||||||
tempPath: D://file
|
tempPath: D://file
|
||||||
|
|||||||
38
kafka-event-data/.gitignore
vendored
Normal file
38
kafka-event-data/.gitignore
vendored
Normal file
@@ -0,0 +1,38 @@
|
|||||||
|
target/
|
||||||
|
!.mvn/wrapper/maven-wrapper.jar
|
||||||
|
!**/src/main/**/target/
|
||||||
|
!**/src/test/**/target/
|
||||||
|
|
||||||
|
### IntelliJ IDEA ###
|
||||||
|
.idea/modules.xml
|
||||||
|
.idea/jarRepositories.xml
|
||||||
|
.idea/compiler.xml
|
||||||
|
.idea/libraries/
|
||||||
|
*.iws
|
||||||
|
*.iml
|
||||||
|
*.ipr
|
||||||
|
|
||||||
|
### Eclipse ###
|
||||||
|
.apt_generated
|
||||||
|
.classpath
|
||||||
|
.factorypath
|
||||||
|
.project
|
||||||
|
.settings
|
||||||
|
.springBeans
|
||||||
|
.sts4-cache
|
||||||
|
|
||||||
|
### NetBeans ###
|
||||||
|
/nbproject/private/
|
||||||
|
/nbbuild/
|
||||||
|
/dist/
|
||||||
|
/nbdist/
|
||||||
|
/.nb-gradle/
|
||||||
|
build/
|
||||||
|
!**/src/main/**/build/
|
||||||
|
!**/src/test/**/build/
|
||||||
|
|
||||||
|
### VS Code ###
|
||||||
|
.vscode/
|
||||||
|
|
||||||
|
### Mac OS ###
|
||||||
|
.DS_Store
|
||||||
153
kafka-event-data/pom.xml
Normal file
153
kafka-event-data/pom.xml
Normal file
@@ -0,0 +1,153 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||||
|
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||||
|
<modelVersion>4.0.0</modelVersion>
|
||||||
|
|
||||||
|
<groupId>com.njcn</groupId>
|
||||||
|
<artifactId>kafka-event-data</artifactId>
|
||||||
|
|
||||||
|
|
||||||
|
<parent>
|
||||||
|
<groupId>com.njcn.product</groupId>
|
||||||
|
<artifactId>CN_Product</artifactId>
|
||||||
|
<version>1.0.0</version>
|
||||||
|
</parent>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.njcn</groupId>
|
||||||
|
<artifactId>njcn-common</artifactId>
|
||||||
|
<version>0.0.1</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
<!-- <dependency>-->
|
||||||
|
<!-- <groupId>com.njcn</groupId>-->
|
||||||
|
<!-- <artifactId>common-redis</artifactId>-->
|
||||||
|
<!-- <version>1.0.0</version>-->
|
||||||
|
<!-- </dependency>-->
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-data-redis</artifactId>
|
||||||
|
<exclusions>
|
||||||
|
<exclusion>
|
||||||
|
<groupId>io.lettuce</groupId>
|
||||||
|
<artifactId>lettuce-core</artifactId>
|
||||||
|
</exclusion>
|
||||||
|
</exclusions>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>redis.clients</groupId>
|
||||||
|
<artifactId>jedis</artifactId>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-websocket</artifactId>
|
||||||
|
<version>2.7.12</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
<!-- 多数据源切换，当数据源为oracle时需要使用 -->
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.baomidou</groupId>
|
||||||
|
<artifactId>dynamic-datasource-spring-boot-starter</artifactId>
|
||||||
|
<version>3.5.1</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.njcn</groupId>
|
||||||
|
<artifactId>spingboot2.3.12</artifactId>
|
||||||
|
<version>2.3.12</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.njcn</groupId>
|
||||||
|
<artifactId>mybatis-plus</artifactId>
|
||||||
|
<version>0.0.1</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
<!--oracle驱动-->
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.oracle.database.jdbc</groupId>
|
||||||
|
<artifactId>ojdbc8</artifactId>
|
||||||
|
<version>21.6.0.0</version>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.oracle.database.nls</groupId>
|
||||||
|
<artifactId>orai18n</artifactId>
|
||||||
|
<version>21.1.0.0</version> <!-- 版本号需要与你的ojdbc版本匹配 -->
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
<!-- Spring Security -->
|
||||||
|
<!-- <dependency>-->
|
||||||
|
<!-- <groupId>org.springframework.boot</groupId>-->
|
||||||
|
<!-- <artifactId>spring-boot-starter-security</artifactId>-->
|
||||||
|
<!-- </dependency>-->
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.kafka</groupId>
|
||||||
|
<artifactId>spring-kafka</artifactId>
|
||||||
|
<version>2.5.14.RELEASE</version> <!-- 与Spring Boot 2.3.x兼容 -->
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.google.guava</groupId>
|
||||||
|
<artifactId>guava</artifactId>
|
||||||
|
<version>32.1.3-jre</version> <!-- 使用最新稳定版 -->
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
</dependencies>
|
||||||
|
|
||||||
|
|
||||||
|
<build>
|
||||||
|
<finalName>kafka-event-data</finalName>
|
||||||
|
<plugins>
|
||||||
|
<plugin>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||||
|
<executions>
|
||||||
|
<execution>
|
||||||
|
<phase>package</phase>
|
||||||
|
<goals>
|
||||||
|
<goal>repackage</goal>
|
||||||
|
</goals>
|
||||||
|
</execution>
|
||||||
|
</executions>
|
||||||
|
</plugin>
|
||||||
|
|
||||||
|
<plugin>
|
||||||
|
<groupId>org.apache.maven.plugins</groupId>
|
||||||
|
<artifactId>maven-compiler-plugin</artifactId>
|
||||||
|
<version>3.8.1</version>
|
||||||
|
<configuration>
|
||||||
|
<source>1.8</source>
|
||||||
|
<target>1.8</target>
|
||||||
|
</configuration>
|
||||||
|
</plugin>
|
||||||
|
</plugins>
|
||||||
|
<resources>
|
||||||
|
<resource>
|
||||||
|
<directory>src/main/resources</directory>
|
||||||
|
<includes>
|
||||||
|
<include>**/*</include>
|
||||||
|
</includes>
|
||||||
|
</resource>
|
||||||
|
<resource>
|
||||||
|
<directory>src/main/java</directory>
|
||||||
|
<includes>
|
||||||
|
<include>**/*.xml</include>
|
||||||
|
</includes>
|
||||||
|
</resource>
|
||||||
|
</resources>
|
||||||
|
|
||||||
|
</build>
|
||||||
|
|
||||||
|
</project>
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
package com.njcn.kafka.event;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.mybatis.spring.annotation.MapperScan;
|
||||||
|
import org.springframework.boot.SpringApplication;
|
||||||
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
@SpringBootApplication(scanBasePackages = "com.njcn")
|
||||||
|
@MapperScan("com.njcn.**.mapper")
|
||||||
|
public class KafkaEventtApplication {
|
||||||
|
|
||||||
|
public static void main(String[] args) {
|
||||||
|
SpringApplication.run(KafkaEventtApplication.class, args);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,29 @@
|
|||||||
|
package com.njcn.kafka.event.controller;
|
||||||
|
|
||||||
|
import com.njcn.kafka.event.pojo.param.PushParam;
|
||||||
|
import com.njcn.kafka.event.service.EventDataService;
|
||||||
|
import lombok.AllArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.web.bind.annotation.PostMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestBody;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/27 下午 2:56【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Slf4j
|
||||||
|
@RestController
|
||||||
|
@RequestMapping("/eventdata")
|
||||||
|
@AllArgsConstructor
|
||||||
|
public class EventDataController {
|
||||||
|
private final EventDataService eventDataService;
|
||||||
|
@PostMapping("/push")
|
||||||
|
public void push(@RequestBody PushParam pushParam) throws Exception {
|
||||||
|
eventDataService.push(pushParam);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,31 @@
|
|||||||
|
package com.njcn.kafka.event.controller;
|
||||||
|
|
||||||
|
import com.njcn.kafka.event.service.KafkaProducerService;
|
||||||
|
import lombok.AllArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestParam;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/27 下午 2:06【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
@RestController
|
||||||
|
@RequestMapping("/test")
|
||||||
|
@AllArgsConstructor
|
||||||
|
public class KafkaTestController {
|
||||||
|
private final KafkaProducerService kafkaProducerService;
|
||||||
|
|
||||||
|
@GetMapping("/testkafka")
|
||||||
|
|
||||||
|
public void test(@RequestParam("context") String context) {
|
||||||
|
kafkaProducerService.sendMessageWithCallback("eventdata",context);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,20 @@
|
|||||||
|
package com.njcn.kafka.event.mapper;
|
||||||
|
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||||
|
import com.njcn.kafka.event.pojo.po.OracleRmpEventDetailPO;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* data-migration
|
||||||
|
*
|
||||||
|
* @author cdf
|
||||||
|
* @date 2024/2/19
|
||||||
|
*/
|
||||||
|
public interface OracleRmpEventDetailPOMapper extends BaseMapper<OracleRmpEventDetailPO> {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,27 @@
|
|||||||
|
package com.njcn.kafka.event.mapper;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||||
|
|
||||||
|
import com.njcn.kafka.event.pojo.dto.LedgerBaseInfoDTO;
|
||||||
|
import com.njcn.kafka.event.pojo.po.PqLine;
|
||||||
|
import org.apache.ibatis.annotations.Param;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
*
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/06/19 下午 1:43【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
public interface PqLineMapper extends BaseMapper<PqLine> {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
List<LedgerBaseInfoDTO> getBaseLedger(@Param("ids")List<Integer> ids, @Param("searchValue")String searchValue);
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,50 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
|
||||||
|
<mapper namespace="com.njcn.kafka.event.mapper.PqLineMapper">
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
<select id="getBaseLedger" resultType="com.njcn.kafka.event.pojo.dto.LedgerBaseInfoDTO">
|
||||||
|
select
|
||||||
|
PQ_GDINFORMATION.name gdName,
|
||||||
|
pq_line.GD_INDEX gdIndex,
|
||||||
|
pq_line.line_index lineId,
|
||||||
|
pq_line.name lineName,
|
||||||
|
PQ_SUBVOLTAGE.SUBV_INDEX busBarId,
|
||||||
|
PQ_SUBVOLTAGE.name busBarName,
|
||||||
|
PQ_SUBVOLTAGE.SCALE,
|
||||||
|
pq_device.dev_index devId,
|
||||||
|
pq_device.name devName,
|
||||||
|
pq_device.ip ip,
|
||||||
|
pq_line.pt1/pq_line.pt2 pt,
|
||||||
|
pq_line.ct1/pq_line.ct2 ct,
|
||||||
|
pq_device.Status runFlag,
|
||||||
|
PQ_SUBSTATION.sub_index stationId,
|
||||||
|
PQ_SUBSTATION.name stationName
|
||||||
|
from
|
||||||
|
pq_line inner JOIN PQ_SUBVOLTAGE on pq_line.SUBV_INDEX = PQ_SUBVOLTAGE.SUBV_INDEX
|
||||||
|
inner JOIN pq_device on PQ_SUBVOLTAGE.DEV_INDEX = pq_device.DEV_INDEX
|
||||||
|
inner JOIN PQ_SUBSTATION on pq_device.SUB_INDEX = PQ_SUBSTATION.SUB_INDEX
|
||||||
|
inner JOIN PQ_GDINFORMATION on pq_line.GD_INDEX =PQ_GDINFORMATION.GD_INDEX
|
||||||
|
<where>
|
||||||
|
and pq_line.line_index in
|
||||||
|
<foreach collection="ids" item="item" open="(" close=")" separator=",">
|
||||||
|
#{item}
|
||||||
|
</foreach>
|
||||||
|
<if test="searchValue!=null and searchValue!=''">
|
||||||
|
and (
|
||||||
|
PQ_GDINFORMATION.name LIKE '%' || #{searchValue} || '%'
|
||||||
|
or PQ_SUBSTATION.name LIKE '%' || #{searchValue} || '%'
|
||||||
|
or pq_line.name LIKE '%' || #{searchValue} || '%'
|
||||||
|
)
|
||||||
|
</if>
|
||||||
|
</where>
|
||||||
|
</select>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
</mapper>
|
||||||
@@ -0,0 +1,68 @@
|
|||||||
|
package com.njcn.kafka.event.pojo.dto;
|
||||||
|
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
import java.math.BigDecimal;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2026/06/04 上午 9:14【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Data
|
||||||
|
public class EventPushDTO {
|
||||||
|
/**
|
||||||
|
* type :
|
||||||
|
* time : 2025-10-09 13:25:14
|
||||||
|
* body : [{"eventdetail_index":"d4963f8f-201f-443f-b293-dda8ef870ee6","lineid":"1591","timeid":"2025-10-09 13:25:14","ms":"509","describe":"","wavetype":"1","persisttime":"23.000","eventvalue":"0.6","eventreason":"","eventtype":"","gdname":"","bdname":"110kV皇后店变","busname":"XX名称","phasicType":"A","pointname":"232待用","wavePath":"192.168.1.102/3_20240515_163022_349"}]
|
||||||
|
*/
|
||||||
|
|
||||||
|
private String type;
|
||||||
|
private LocalDateTime time;
|
||||||
|
private List<EventDTO> body;
|
||||||
|
|
||||||
|
@Data
|
||||||
|
public static class EventDTO {
|
||||||
|
/**
|
||||||
|
* eventdetail_index : d4963f8f-201f-443f-b293-dda8ef870ee6
|
||||||
|
* lineid : 1591
|
||||||
|
* timeid : 2025-10-09 13:25:14
|
||||||
|
* ms : 509
|
||||||
|
* describe :
|
||||||
|
* wavetype : 1
|
||||||
|
* persisttime : 23.000
|
||||||
|
* eventvalue : 0.6
|
||||||
|
* eventreason :
|
||||||
|
* eventtype :
|
||||||
|
* gdname :
|
||||||
|
* bdname : 110kV皇后店变
|
||||||
|
* busname : XX名称
|
||||||
|
* phasicType : A
|
||||||
|
* pointname : 232待用
|
||||||
|
* wavePath : 192.168.1.102/3_20240515_163022_349
|
||||||
|
*/
|
||||||
|
|
||||||
|
private String eventdetail_index;
|
||||||
|
private Integer lineid;
|
||||||
|
private LocalDateTime timeid;
|
||||||
|
private Integer ms;
|
||||||
|
private String describe;
|
||||||
|
private String wavetype;
|
||||||
|
private BigDecimal persisttime;
|
||||||
|
private BigDecimal eventvalue;
|
||||||
|
private String eventreason;
|
||||||
|
private String eventtype;
|
||||||
|
private String gdname;
|
||||||
|
private String bdname;
|
||||||
|
private String busname;
|
||||||
|
private String phasicType;
|
||||||
|
private String pointname;
|
||||||
|
private String wavePath;
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,43 @@
|
|||||||
|
package com.njcn.kafka.event.pojo.dto;
|
||||||
|
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @Author: cdf
|
||||||
|
* @CreateTime: 2025-06-25
|
||||||
|
* @Description:
|
||||||
|
*/
|
||||||
|
@Data
|
||||||
|
public class LedgerBaseInfoDTO {
|
||||||
|
private String gdName;
|
||||||
|
private String gdIndex;
|
||||||
|
|
||||||
|
private Integer lineId;
|
||||||
|
|
||||||
|
private String lineName;
|
||||||
|
|
||||||
|
private Integer busBarId;
|
||||||
|
|
||||||
|
private String busBarName;
|
||||||
|
|
||||||
|
private String scale;
|
||||||
|
|
||||||
|
private Integer devId;
|
||||||
|
|
||||||
|
private String devName;
|
||||||
|
|
||||||
|
private String objName;
|
||||||
|
|
||||||
|
private Integer stationId;
|
||||||
|
|
||||||
|
private String stationName;
|
||||||
|
//通讯状态
|
||||||
|
private Integer runFlag=0;
|
||||||
|
|
||||||
|
private Integer eventCount;
|
||||||
|
private String ip;
|
||||||
|
|
||||||
|
private int pt;
|
||||||
|
private int ct;
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,34 @@
|
|||||||
|
package com.njcn.kafka.event.pojo.param;
|
||||||
|
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/28 上午 8:47【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Data
|
||||||
|
public class PushParam {
|
||||||
|
|
||||||
|
//补招起始日期_yyyy-MM-dd(按小时跑的任务可加时分秒
|
||||||
|
private LocalDateTime beginTime;
|
||||||
|
|
||||||
|
//补招截止日期_yyyy-MM-dd(按小时跑的任务可加时分秒)
|
||||||
|
private LocalDateTime endTime;
|
||||||
|
|
||||||
|
//时间日期_yyyy-MM-dd(按小时跑的任务可加时分秒)
|
||||||
|
private String dataDate;
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 待计算的对象索引集合，监测点、设备、母线、变电站、单位等等
|
||||||
|
*/
|
||||||
|
private List<String> idList;
|
||||||
|
|
||||||
|
}
|
||||||
@@ -0,0 +1,174 @@
|
|||||||
|
package com.njcn.kafka.event.pojo.po;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.annotation.IdType;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableField;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableId;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableName;
|
||||||
|
import lombok.Data;
|
||||||
|
import lombok.EqualsAndHashCode;
|
||||||
|
|
||||||
|
import java.io.Serializable;
|
||||||
|
import java.math.BigDecimal;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.Date;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* data-migration
|
||||||
|
*
|
||||||
|
* @author cdf
|
||||||
|
* @date 2024/2/19
|
||||||
|
*/
|
||||||
|
@Data
|
||||||
|
@EqualsAndHashCode(callSuper = false)
|
||||||
|
@TableName("PQS_EVENTDETAIL")
|
||||||
|
public class OracleRmpEventDetailPO implements Serializable {
|
||||||
|
private static final long serialVersionUID = 1L;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 暂时事件ID
|
||||||
|
*/
|
||||||
|
@TableId(value = "EVENTDETAIL_INDEX",type = IdType.ASSIGN_ID)
|
||||||
|
private String eventdetailIndex;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 监测点ID
|
||||||
|
*/
|
||||||
|
@TableField(value = "LINEID")
|
||||||
|
private Integer lineid;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 统计指标类型
|
||||||
|
*/
|
||||||
|
@TableField(value = "MS")
|
||||||
|
private Integer ms;
|
||||||
|
/**
|
||||||
|
* 统计指标类型
|
||||||
|
*/
|
||||||
|
@TableField(value = "WAVETYPE")
|
||||||
|
private String wavetype;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 暂降原因（字典表PQS_Dicdata）
|
||||||
|
*/
|
||||||
|
@TableField(value = "EVENTREASON")
|
||||||
|
private String eventreason;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 暂降类型（字典表PQS_Dicdata）
|
||||||
|
*/
|
||||||
|
@TableField(value = "EVENTTYPE")
|
||||||
|
private String eventtype;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 事件关联分析表Guid
|
||||||
|
*/
|
||||||
|
@TableField(value = "EVENTASS_INDEX")
|
||||||
|
private String eventassIndex;
|
||||||
|
|
||||||
|
@TableField(value = "DQTIME")
|
||||||
|
private Double dqTime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 特征值计算更新时间（外键PQS_Relevance的Time字段）
|
||||||
|
*/
|
||||||
|
@TableField(value = "DEALTIME")
|
||||||
|
private Date dealTime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 默认事件个数为0
|
||||||
|
*/
|
||||||
|
@TableField(value = "NUM")
|
||||||
|
private Integer num;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 波形文件是否从装置招到本地(0：未招，1：已招)默认值为0
|
||||||
|
*/
|
||||||
|
@TableField(value = "FILEFLAG")
|
||||||
|
private Integer fileFlag;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 特征值计算标志（0，未处理；1，已处理; 2，已处理，无结果;3，计算失败）默认值为0
|
||||||
|
*/
|
||||||
|
@TableField(value = "DEALFLAG")
|
||||||
|
private Integer dealFlag;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 处理结果第一条事件发生时间(读comtra文件获取)
|
||||||
|
*/
|
||||||
|
@TableField(value = "FIRSTTIME")
|
||||||
|
private LocalDateTime firstTime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 处理结果第一条事件暂降类型（字典表PQS_Dicdata）
|
||||||
|
*/
|
||||||
|
@TableField(value = "FIRSTTYPE")
|
||||||
|
private String firstType;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 处理结果第一条事件发生时间毫秒(读comtra文件获取)
|
||||||
|
*/
|
||||||
|
@TableField(value = "FIRSTMS")
|
||||||
|
private BigDecimal firstMs;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 暂降能量
|
||||||
|
*/
|
||||||
|
@TableField(value = "ENERGY")
|
||||||
|
private Double energy;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 暂降严重度
|
||||||
|
*/
|
||||||
|
@TableField(value = "SEVERITY")
|
||||||
|
private Double severity;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 暂降源与监测位置关系 Upper:上游；Lower :下游；Unknown :未知;为空则是未计算
|
||||||
|
*/
|
||||||
|
@TableField(value = "SAGSOURCE")
|
||||||
|
private String sagsource;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 开始时间
|
||||||
|
*/
|
||||||
|
@TableField(value = "TIMEID")
|
||||||
|
private LocalDateTime timeid;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 持续时间，单位秒
|
||||||
|
*/
|
||||||
|
@TableField(value = "PERSISTTIME")
|
||||||
|
private BigDecimal persisttime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 特征幅值
|
||||||
|
*/
|
||||||
|
@TableField(value = "EVENTVALUE")
|
||||||
|
private BigDecimal eventvalue;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 相别
|
||||||
|
*/
|
||||||
|
@TableField(value = "PHASIC_TYPE")
|
||||||
|
private String phasicType;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 事件描述
|
||||||
|
*/
|
||||||
|
@TableField(value = "DESCRIBE")
|
||||||
|
private String describe;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 波形路径
|
||||||
|
*/
|
||||||
|
@TableField(value = "WAVENAME")
|
||||||
|
private String wavePath;
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@TableField(value = "TRANSIENTVALUE")
|
||||||
|
private Double transientValue;
|
||||||
|
|
||||||
|
@TableField(value = "CREATE_TIME")
|
||||||
|
private LocalDateTime createTime;
|
||||||
|
}
|
||||||
@@ -0,0 +1,133 @@
|
|||||||
|
package com.njcn.kafka.event.pojo.po;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.annotation.IdType;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableField;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableId;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableName;
|
||||||
|
import lombok.Data;
|
||||||
|
import lombok.NoArgsConstructor;
|
||||||
|
|
||||||
|
/**
|
||||||
|
*
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/06/19 下午 1:43【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠?
|
||||||
|
*/
|
||||||
|
@Data
|
||||||
|
@NoArgsConstructor
|
||||||
|
@TableName(value = "PQ_LINE")
|
||||||
|
public class PqLine {
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableId(value = "LINE_INDEX", type = IdType.INPUT)
|
||||||
|
private Integer lineIndex;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "GD_INDEX")
|
||||||
|
private Integer gdIndex;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠?
|
||||||
|
*/
|
||||||
|
@TableField(value = "SUB_INDEX")
|
||||||
|
private Integer subIndex;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "SUBV_INDEX")
|
||||||
|
private Integer subvIndex;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "DEV_INDEX")
|
||||||
|
private Integer devIndex;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "\"NAME\"")
|
||||||
|
private String name;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* PT靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "PT1")
|
||||||
|
private Double pt1;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* PT靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "PT2")
|
||||||
|
private Double pt2;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* CT靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "CT1")
|
||||||
|
private Double ct1;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* CT靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "CT2")
|
||||||
|
private Double ct2;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "DEVCMP")
|
||||||
|
private Double devcmp;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "DLCMP")
|
||||||
|
private Double dlcmp;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "JZCMP")
|
||||||
|
private Double jzcmp;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "XYCMP")
|
||||||
|
private Double xycmp;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠?靠靠靠靠靠靠?
|
||||||
|
*/
|
||||||
|
@TableField(value = "SUBV_NO")
|
||||||
|
private Integer subvNo;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* (靠PQS_Dictionary?靠靠Guid
|
||||||
|
*/
|
||||||
|
@TableField(value = "\"SCALE\"")
|
||||||
|
private String scale;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 靠靠
|
||||||
|
*/
|
||||||
|
@TableField(value = "SUBV_NAME")
|
||||||
|
private String subvName;
|
||||||
|
|
||||||
|
@TableField(exist = false)
|
||||||
|
private String subName;
|
||||||
|
|
||||||
|
@TableField(exist = false)
|
||||||
|
private String deptName;
|
||||||
|
}
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
package com.njcn.kafka.event.service;
|
||||||
|
|
||||||
|
|
||||||
|
import com.njcn.kafka.event.pojo.param.PushParam;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/28 上午 8:38【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
public interface EventDataService {
|
||||||
|
void push(PushParam pushParam) throws Exception;
|
||||||
|
}
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
package com.njcn.kafka.event.service;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/27 下午 1:57【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
public interface KafkaProducerService {
|
||||||
|
void sendMessage(String topic, String message);
|
||||||
|
|
||||||
|
// 带回调的发送
|
||||||
|
void sendMessageWithCallback(String topic, String message);
|
||||||
|
}
|
||||||
@@ -0,0 +1,74 @@
|
|||||||
|
package com.njcn.kafka.event.service.impl;
|
||||||
|
|
||||||
|
|
||||||
|
import cn.hutool.json.JSONUtil;
|
||||||
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||||
|
import com.njcn.kafka.event.mapper.OracleRmpEventDetailPOMapper;
|
||||||
|
import com.njcn.kafka.event.mapper.PqLineMapper;
|
||||||
|
import com.njcn.kafka.event.pojo.dto.EventPushDTO;
|
||||||
|
import com.njcn.kafka.event.pojo.dto.LedgerBaseInfoDTO;
|
||||||
|
import com.njcn.kafka.event.pojo.param.PushParam;
|
||||||
|
import com.njcn.kafka.event.pojo.po.OracleRmpEventDetailPO;
|
||||||
|
import com.njcn.kafka.event.service.EventDataService;
|
||||||
|
import com.njcn.kafka.event.service.KafkaProducerService;
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import org.springframework.beans.BeanUtils;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
import org.springframework.util.CollectionUtils;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.function.Function;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/28 上午 8:38【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Service
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
public class EventDataServiceImpl implements EventDataService {
|
||||||
|
|
||||||
|
|
||||||
|
private final KafkaProducerService kafkaProducerService;
|
||||||
|
private final OracleRmpEventDetailPOMapper oracleRmpEventDetailPOMapper;
|
||||||
|
private final PqLineMapper pqLineMapper;
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void push(PushParam pushParam) throws Exception {
|
||||||
|
LocalDateTime beginTime = pushParam.getBeginTime();
|
||||||
|
LocalDateTime endTime = pushParam.getEndTime();
|
||||||
|
List<OracleRmpEventDetailPO> oracleRmpEventDetailPOList = oracleRmpEventDetailPOMapper.selectList(
|
||||||
|
new LambdaQueryWrapper<OracleRmpEventDetailPO>()
|
||||||
|
.between(OracleRmpEventDetailPO::getCreateTime,beginTime,endTime));
|
||||||
|
if(!CollectionUtils.isEmpty(oracleRmpEventDetailPOList)){
|
||||||
|
List<Integer> lineIds = oracleRmpEventDetailPOList.stream().map(OracleRmpEventDetailPO::getLineid).collect(Collectors.toList());
|
||||||
|
List<LedgerBaseInfoDTO> baseLedger = pqLineMapper.getBaseLedger(lineIds, null);
|
||||||
|
Map<Integer, LedgerBaseInfoDTO> ledgerBaseInfoDTOMap = baseLedger.stream().collect(Collectors.toMap(LedgerBaseInfoDTO::getLineId, Function.identity()));
|
||||||
|
EventPushDTO eventPushDTO = new EventPushDTO();
|
||||||
|
eventPushDTO.setTime(LocalDateTime.now());
|
||||||
|
List<EventPushDTO.EventDTO> collect = oracleRmpEventDetailPOList.stream().map(temp -> {
|
||||||
|
EventPushDTO.EventDTO eventDTO = new EventPushDTO.EventDTO();
|
||||||
|
BeanUtils.copyProperties(temp,eventDTO);
|
||||||
|
eventDTO.setEventdetail_index(temp.getEventdetailIndex());
|
||||||
|
eventDTO.setBusname(ledgerBaseInfoDTOMap.get(temp.getLineid()).getBusBarName());
|
||||||
|
eventDTO.setBdname(ledgerBaseInfoDTOMap.get(temp.getLineid()).getStationName());
|
||||||
|
eventDTO.setPointname(ledgerBaseInfoDTOMap.get(temp.getLineid()).getLineName());
|
||||||
|
eventDTO.setGdname(ledgerBaseInfoDTOMap.get(temp.getLineid()).getGdName());
|
||||||
|
return eventDTO;
|
||||||
|
}).collect(Collectors.toList());
|
||||||
|
eventPushDTO.setBody(collect);
|
||||||
|
String jsonStr = JSONUtil.toJsonStr(collect);
|
||||||
|
|
||||||
|
kafkaProducerService.sendMessage("eventdata",jsonStr);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,50 @@
|
|||||||
|
package com.njcn.kafka.event.service.impl;
|
||||||
|
|
||||||
|
import com.njcn.kafka.event.service.KafkaProducerService;
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import org.springframework.kafka.core.KafkaTemplate;
|
||||||
|
import org.springframework.kafka.support.SendResult;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
import org.springframework.util.concurrent.ListenableFuture;
|
||||||
|
import org.springframework.util.concurrent.ListenableFutureCallback;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Description:
|
||||||
|
* Date: 2025/05/27 下午 2:00【需求编号】
|
||||||
|
*
|
||||||
|
* @author clam
|
||||||
|
* @version V1.0.0
|
||||||
|
*/
|
||||||
|
@Service
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
public class KafkaProducerServiceImpl implements KafkaProducerService {
|
||||||
|
|
||||||
|
private final KafkaTemplate<String, String> kafkaTemplate;
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void sendMessage(String topic, String message) {
|
||||||
|
kafkaTemplate.send(topic, message);
|
||||||
|
}
|
||||||
|
@Override
|
||||||
|
// 带回调的发送
|
||||||
|
public void sendMessageWithCallback(String topic, String message) {
|
||||||
|
ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
|
||||||
|
|
||||||
|
future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onSuccess(SendResult<String, String> result) {
|
||||||
|
System.out.println("Sent message=[" + message +
|
||||||
|
"] with offset=[" + result.getRecordMetadata().offset() + "]");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onFailure(Throwable ex) {
|
||||||
|
System.out.println("Unable to send message=[" +
|
||||||
|
message + "] due to : " + ex.getMessage());
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
88
kafka-event-data/src/main/resources/application-dev.yml
Normal file
88
kafka-event-data/src/main/resources/application-dev.yml
Normal file
@@ -0,0 +1,88 @@
|
|||||||
|
spring:
|
||||||
|
application:
|
||||||
|
name: kafka-event-data
|
||||||
|
kafka:
|
||||||
|
# Kafka 服务器地址
|
||||||
|
bootstrap-servers: 192.168.2.130:9092
|
||||||
|
|
||||||
|
# 生产者配置
|
||||||
|
producer:
|
||||||
|
key-serializer: org.apache.kafka.common.serialization.StringSerializer
|
||||||
|
value-serializer: org.apache.kafka.common.serialization.StringSerializer
|
||||||
|
# 可选配置
|
||||||
|
acks: all
|
||||||
|
retries: 3
|
||||||
|
batch-size: 16384
|
||||||
|
buffer-memory: 33554432
|
||||||
|
|
||||||
|
# 消费者配置
|
||||||
|
consumer:
|
||||||
|
group-id: my-group
|
||||||
|
auto-offset-reset: earliest
|
||||||
|
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
|
||||||
|
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
|
||||||
|
# 可选配置
|
||||||
|
enable-auto-commit: false
|
||||||
|
max-poll-records: 500
|
||||||
|
datasource:
|
||||||
|
dynamic:
|
||||||
|
primary: master
|
||||||
|
strict: false # 是否严格匹配数据源，默认false
|
||||||
|
druid: # 如果使用Druid连接池
|
||||||
|
validation-query: SELECT 1 FROM DUAL # 达梦专用校验SQL
|
||||||
|
initial-size: 10
|
||||||
|
# 初始化大小，最小，最大
|
||||||
|
min-idle: 20
|
||||||
|
maxActive: 500
|
||||||
|
# 配置获取连接等待超时的时间
|
||||||
|
maxWait: 60000
|
||||||
|
# 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
|
||||||
|
timeBetweenEvictionRunsMillis: 60000
|
||||||
|
# 配置一个连接在池中最小生存的时间，单位是毫秒
|
||||||
|
minEvictableIdleTimeMillis: 300000
|
||||||
|
testWhileIdle: true
|
||||||
|
testOnBorrow: true
|
||||||
|
testOnReturn: false
|
||||||
|
# 打开PSCache，并且指定每个连接上PSCache的大小
|
||||||
|
poolPreparedStatements: true
|
||||||
|
maxPoolPreparedStatementPerConnectionSize: 20
|
||||||
|
datasource:
|
||||||
|
master:
|
||||||
|
url: jdbc:oracle:thin:@192.168.1.51:1521:pqsbase
|
||||||
|
username: pqsadmin_hn
|
||||||
|
password: pqsadmin
|
||||||
|
driver-class-name: oracle.jdbc.OracleDriver
|
||||||
|
# salve:
|
||||||
|
# driver-class-name: dm.jdbc.driver.DmDriver
|
||||||
|
# url: jdbc:dm://192.168.1.21:5236/PQSADMIN?useUnicode=true&characterEncoding=utf-8
|
||||||
|
# username: PQSADMINLN
|
||||||
|
# password: Pqsadmin123
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
redis:
|
||||||
|
database: 10
|
||||||
|
host: localhost
|
||||||
|
port: 6379
|
||||||
|
timeout: 5000
|
||||||
|
jedis:
|
||||||
|
pool:
|
||||||
|
max-active: 20
|
||||||
|
max-wait: 8000
|
||||||
|
max-idle: 8
|
||||||
|
min-idle: 2
|
||||||
|
test-on-borrow: true # 借出连接时验证
|
||||||
|
test-on-return: true # 归还连接时验证
|
||||||
|
test-while-idle: true # 空闲时验证
|
||||||
|
|
||||||
|
smsServer:
|
||||||
|
info: http://22.33.194.50:18096
|
||||||
|
netInfo: http://22.33.191.206:18096
|
||||||
|
account: xbjbpt
|
||||||
|
password: WLv8w071
|
||||||
|
aliyun:
|
||||||
|
oss:
|
||||||
|
endpoint: oss-cn-beijing.aliyuncs.com
|
||||||
|
accessKeyId: LTAI5tQYuyu1PpiCdeM74PT6
|
||||||
|
accessKeySecret: vTGHcQOCF9u7w9FL3HAHJO1oufVWru
|
||||||
|
bucketName: cn-comtrade
|
||||||
83
kafka-event-data/src/main/resources/application.yml
Normal file
83
kafka-event-data/src/main/resources/application.yml
Normal file
@@ -0,0 +1,83 @@
|
|||||||
|
#当前服务的基本信息
|
||||||
|
microservice:
|
||||||
|
ename: 12345
|
||||||
|
name: 12345
|
||||||
|
server:
|
||||||
|
port: 18094
|
||||||
|
spring:
|
||||||
|
application:
|
||||||
|
name: kafka-event-data
|
||||||
|
profiles:
|
||||||
|
active: dev
|
||||||
|
|
||||||
|
|
||||||
|
#mybatis配置信息
|
||||||
|
mybatis-plus:
|
||||||
|
mapper-locations: classpath*:com/njcn/**/mapping/*.xml
|
||||||
|
#别名扫描
|
||||||
|
type-aliases-package: com.njcn.product.event.**.pojo
|
||||||
|
configuration:
|
||||||
|
#驼峰命名
|
||||||
|
map-underscore-to-camel-case: true
|
||||||
|
#配置sql日志输出
|
||||||
|
log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
|
||||||
|
#关闭日志输出
|
||||||
|
# log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
|
||||||
|
global-config:
|
||||||
|
db-config:
|
||||||
|
#指定主键生成策略
|
||||||
|
id-type: assign_uuid
|
||||||
|
|
||||||
|
|
||||||
|
SYS_TYPE_ZT: 1cfcd6e2-c5fe-4b15-988a-32b90f1170c1
|
||||||
|
SYS_TYPE_WT: 983f9dfe-4f9a-4c96-89d8-7d425a1f1d6c
|
||||||
|
db:
|
||||||
|
type: oracle
|
||||||
|
|
||||||
|
#文件位置配置
|
||||||
|
business:
|
||||||
|
#处理波形数据位置
|
||||||
|
wavePath: D://Comtrade
|
||||||
|
targetPath: /pqmonitor
|
||||||
|
exportBaseDir: D://exportComtrade
|
||||||
|
eventCronExpression: 0 0/2 * * * ?
|
||||||
|
failCronExpression: 0 5/10 * * * ?
|
||||||
|
userCronExpression: 0 5 1 * * ?
|
||||||
|
sendMessageCronExpression : 0 */3 * * * ?
|
||||||
|
syncinterval: 2
|
||||||
|
failsyncinterval: 1440
|
||||||
|
#实时短信功能
|
||||||
|
RealTimeSMSSwitch: false
|
||||||
|
#wavePath: /usr/local/comtrade
|
||||||
|
#处理临时数据
|
||||||
|
tempPath: D://file
|
||||||
|
#tempPath: /usr/local/file
|
||||||
|
#文件存储的方式 3.本地存储
|
||||||
|
file:
|
||||||
|
storage: 3
|
||||||
|
#oss服务器配置
|
||||||
|
min:
|
||||||
|
io:
|
||||||
|
endpoint: http://192.168.1.13:9009
|
||||||
|
accessKey: minio
|
||||||
|
secretKey: minio@123
|
||||||
|
bucket: excelreport
|
||||||
|
#华为obs服务器配置
|
||||||
|
huawei:
|
||||||
|
access-key: J9GS9EA79PZ60OK23LWP
|
||||||
|
security-key: BirGrAFDSLxU8ow5fffyXgZRAmMRb1R1AdqCI60d
|
||||||
|
obs:
|
||||||
|
bucket: test-8601
|
||||||
|
endpoint: https://obs.cn-east-3.myhuaweicloud.com
|
||||||
|
# 单位为秒
|
||||||
|
expire: 3600
|
||||||
|
#线程池配置信息
|
||||||
|
threadPool:
|
||||||
|
corePoolSize: 10
|
||||||
|
maxPoolSize: 20
|
||||||
|
queueCapacity: 500
|
||||||
|
keepAliveSeconds: 60
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
145
kafka-event-data/src/main/resources/logback.xml
Normal file
145
kafka-event-data/src/main/resources/logback.xml
Normal file
@@ -0,0 +1,145 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<configuration scan="true" scanPeriod="20 seconds" debug="false">
|
||||||
|
<include resource="org/springframework/boot/logging/logback/defaults.xml" />
|
||||||
|
|
||||||
|
<springProperty scope="context" name="log.projectName" source="spring.application.name" defaultValue="event_msg"/>
|
||||||
|
<springProperty scope="context" name="logCommonLevel" source="log.commonLevel" defaultValue="info"/>
|
||||||
|
<springProperty scope="context" name="logHomeDir" source="log.homeDir" defaultValue="D:\logs"/>
|
||||||
|
|
||||||
|
<conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
|
||||||
|
<conversionRule conversionWord="wex"
|
||||||
|
converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
|
||||||
|
<conversionRule conversionWord="ec"
|
||||||
|
converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
|
||||||
|
|
||||||
|
|
||||||
|
<!--日志输出格式-->
|
||||||
|
<property name="log.pattern" value="|-%d{yyyy-MM-dd HH:mm:ss.SSS} ${LOG_LEVEL_PATTERN:-%level} ${log.projectName} -- %t %logger{100}.%M ==> %m%n${Log_EXCEPTION_CONVERSION_WORD:-%ec}}"/>
|
||||||
|
<property name="log.maxHistory" value="30"/>
|
||||||
|
<!-- 控制台输出（可选） -->
|
||||||
|
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
|
||||||
|
<encoder>
|
||||||
|
<charset>UTF-8</charset> <!-- 控制台也建议指定 -->
|
||||||
|
<pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</pattern>
|
||||||
|
</encoder>
|
||||||
|
</appender>
|
||||||
|
|
||||||
|
<!--客户端输出日志-->
|
||||||
|
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
|
||||||
|
<encoder>
|
||||||
|
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
|
||||||
|
<charset>UTF-8</charset>
|
||||||
|
</encoder>
|
||||||
|
</appender>
|
||||||
|
|
||||||
|
<!--系统中常规的debug日志-->
|
||||||
|
<!-- 滚动记录文件，先将日志记录到指定文件，当符合某个条件时，将日志记录到其他文件 RollingFileAppender -->
|
||||||
|
<appender name="DEBUG" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||||
|
<file>
|
||||||
|
${logHomeDir}/${log.projectName}/debug/debug.log
|
||||||
|
</file>
|
||||||
|
<!-- 如果日志级别等于配置级别，过滤器会根据onMath 和 onMismatch接收或拒绝日志。 -->
|
||||||
|
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||||
|
<!-- 设置过滤级别 -->
|
||||||
|
<level>DEBUG</level>
|
||||||
|
<!-- 用于配置符合过滤条件的操作 -->
|
||||||
|
<onMatch>ACCEPT</onMatch>
|
||||||
|
<!-- 用于配置不符合过滤条件的操作 -->
|
||||||
|
<onMismatch>DENY</onMismatch>
|
||||||
|
</filter>
|
||||||
|
<!-- 最常用的滚动策略，它根据时间来制定滚动策略.既负责滚动也负责触发滚动 SizeAndTimeBasedRollingPolicy-->
|
||||||
|
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
|
||||||
|
<!--日志输出位置 可相对、和绝对路径 -->
|
||||||
|
<fileNamePattern>
|
||||||
|
${logHomeDir}/${log.projectName}/debug/debug.log.%d{yyyy-MM-dd}.%i.log
|
||||||
|
</fileNamePattern>
|
||||||
|
<maxFileSize>10MB</maxFileSize>
|
||||||
|
<!-- 可选节点，控制保留的归档文件的最大数量，超出数量就删除旧文件,假设设置每个月滚动，且<maxHistory>是6，
|
||||||
|
则只保存最近6个月的文件，删除之前的旧文件。注意，删除旧文件是，那些为了归档而创建的目录也会被删除 -->
|
||||||
|
<maxHistory>${log.maxHistory:-30}</maxHistory>
|
||||||
|
<!--重启清理日志文件-->
|
||||||
|
<!-- <cleanHistoryOnStart>true</cleanHistoryOnStart>-->
|
||||||
|
<!--每个文件最多100MB，保留N天的历史记录，但最多20GB-->
|
||||||
|
<!--<totalSizeCap>20GB</totalSizeCap>-->
|
||||||
|
<!--日志文件最大的大小-->
|
||||||
|
<!--<MaxFileSize>${log.maxSize}</MaxFileSize>-->
|
||||||
|
</rollingPolicy>
|
||||||
|
<encoder>
|
||||||
|
<pattern>
|
||||||
|
${log.pattern}
|
||||||
|
</pattern>
|
||||||
|
<charset>UTF-8</charset>
|
||||||
|
</encoder>
|
||||||
|
</appender>
|
||||||
|
|
||||||
|
<!--系统中常规的info日志-->
|
||||||
|
<appender name="INFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||||
|
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||||
|
<level>INFO</level>
|
||||||
|
<onMatch>ACCEPT</onMatch>
|
||||||
|
<onMismatch>DENY</onMismatch>
|
||||||
|
</filter>
|
||||||
|
<file>
|
||||||
|
${logHomeDir}/${log.projectName}/info/info.log
|
||||||
|
</file>
|
||||||
|
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
|
||||||
|
<fileNamePattern>
|
||||||
|
${logHomeDir}/${log.projectName}/info/info.log.%d{yyyy-MM-dd}.%i.log
|
||||||
|
</fileNamePattern>
|
||||||
|
<maxFileSize>10MB</maxFileSize>
|
||||||
|
<maxHistory>${log.maxHistory:-30}</maxHistory>
|
||||||
|
</rollingPolicy>
|
||||||
|
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
|
||||||
|
<pattern>
|
||||||
|
${log.pattern}
|
||||||
|
</pattern>
|
||||||
|
<charset>UTF-8</charset>
|
||||||
|
</encoder>
|
||||||
|
</appender>
|
||||||
|
|
||||||
|
|
||||||
|
<!--系统中常规的error日志-->
|
||||||
|
<appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||||
|
<file>
|
||||||
|
${logHomeDir}/${log.projectName}/error/error.log
|
||||||
|
</file>
|
||||||
|
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||||
|
<level>ERROR</level>
|
||||||
|
<onMatch>ACCEPT</onMatch>
|
||||||
|
<onMismatch>DENY</onMismatch>
|
||||||
|
</filter>
|
||||||
|
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
|
||||||
|
<fileNamePattern>
|
||||||
|
${logHomeDir}/${log.projectName}/error/error.log.%d{yyyy-MM-dd}.%i.log
|
||||||
|
</fileNamePattern>
|
||||||
|
<maxFileSize>10MB</maxFileSize>
|
||||||
|
<maxHistory>${log.maxHistory:-30}</maxHistory>
|
||||||
|
</rollingPolicy>
|
||||||
|
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
|
||||||
|
<pattern>
|
||||||
|
${log.pattern}
|
||||||
|
</pattern>
|
||||||
|
<charset>UTF-8</charset>
|
||||||
|
</encoder>
|
||||||
|
</appender>
|
||||||
|
|
||||||
|
<logger name="org.apache.catalina.startup.DigesterFactory" level="ERROR"/>
|
||||||
|
<logger name="org.apache.catalina.util.LifecycleBase" level="ERROR"/>
|
||||||
|
<logger name="org.apache.coyote.http11.Http11NioProtocol" level="WARN"/>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
<logger name="com.njcn" level="INFO" additivity="false">
|
||||||
|
<appender-ref ref="CONSOLE"/>
|
||||||
|
<appender-ref ref="DEBUG"/>
|
||||||
|
<appender-ref ref="INFO"/>
|
||||||
|
<appender-ref ref="ERROR"/>
|
||||||
|
</logger>
|
||||||
|
|
||||||
|
<root level="${logCommonLevel}">
|
||||||
|
<appender-ref ref="CONSOLE"/>
|
||||||
|
<appender-ref ref="DEBUG"/>
|
||||||
|
<appender-ref ref="INFO"/>
|
||||||
|
<appender-ref ref="ERROR"/>
|
||||||
|
</root>
|
||||||
|
</configuration>
|
||||||
12
kafka-event-data/src/main/resources/templates/hello.html
Normal file
12
kafka-event-data/src/main/resources/templates/hello.html
Normal file
@@ -0,0 +1,12 @@
|
|||||||
|
<!DOCTYPE html>
|
||||||
|
<html xmlns:th="http://www.thymeleaf.org">
|
||||||
|
<head>
|
||||||
|
<meta charset="UTF-8" />
|
||||||
|
<title></title>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
<p>
|
||||||
|
<span th:text="${hello}">Hello!</span>
|
||||||
|
</p>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
Reference in New Issue
Block a user