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

无线系统实时数据功能

Browse Source
This commit is contained in:
xy
2024-10-11 18:18:30 +08:00
parent 68e28880d0
commit cfd395a11c
28 changed files with 868 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
iot-access/access-api/src/main/java/com/njcn/access/api/AskDeviceDataFeignClient.java
View File

@@ -1,8 +1,15 @@
package com.njcn.access.api;
import com.njcn.access.api.fallback.AskDeviceDataClientFallbackFactory;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
@@ -31,4 +38,7 @@ public interface AskDeviceDataFeignClient {
@PostMapping("/deleteFolder")
HttpResult<String> deleteFolder(@RequestParam("nDid") String nDid, @RequestParam("path") String path);
@PostMapping("/askRealData")
HttpResult<String> askRealData(@RequestParam("nDid") String nDid, @RequestParam("idx") Integer idx, @RequestParam("clDId") Integer clDId);
}

6
iot-access/access-api/src/main/java/com/njcn/access/api/fallback/AskDeviceDataClientFallbackFactory.java
View File

@@ -67,6 +67,12 @@ public class AskDeviceDataClientFallbackFactory implements FallbackFactory<AskDe
log.error("{}异常，降级处理，异常为：{}","删除文件夹",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<String> askRealData(String nDid, Integer idx, Integer clDid) {
log.error("{}异常，降级处理，异常为：{}","询问装置实时数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

6
iot-access/access-api/src/main/java/com/njcn/access/pojo/dto/AskDataDto.java
View File

@@ -32,6 +32,12 @@ public class AskDataDto {
@ParamName("EndTime")
private Integer EndTime;
@ParamName("RtDuration")
private Integer RtDuration;
@ParamName("DsNameIdx")
private Integer DsNameIdx;
@ParamName("DataArray")
private DataArrayDto DataArray;
}

8
iot-access/access-boot/pom.xml
View File

@@ -66,8 +66,12 @@
<dependency>
<groupId>com.njcn</groupId>
<artifactId>zl-event-api</artifactId>
<version>1.0.0</version>
<scope>compile</scope>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-device-biz</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>

14
iot-access/access-boot/src/main/java/com/njcn/access/controller/AskDeviceDataController.java
View File

@@ -109,5 +109,19 @@ public class AskDeviceDataController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/askRealData")
@ApiOperation("询问装置实时数据")
@ApiImplicitParams({
@ApiImplicitParam(name = "nDid", value = "装置nDid"),
@ApiImplicitParam(name = "idx", value = "数据集编号"),
@ApiImplicitParam(name = "clDId", value = "逻辑子设备id")
})
public HttpResult<String> askRealData(@RequestParam("nDid") String nDid, @RequestParam("idx") Integer idx, @RequestParam("clDId") Integer clDId){
String methodDescribe = getMethodDescribe("askRealData");
askDeviceDataService.askRealData(nDid,idx,clDId);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
}

17
iot-access/access-boot/src/main/java/com/njcn/access/mapper/OverlimitMapper.java Normal file
View File

@@ -0,0 +1,17 @@
package com.njcn.access.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.njcn.device.biz.pojo.po.Overlimit;
/**
* <p>
* Mapper 接口
* </p>
*
* @author xy
*/
public interface OverlimitMapper extends BaseMapper<Overlimit> {
}

5
iot-access/access-boot/src/main/java/com/njcn/access/service/AskDeviceDataService.java
View File

@@ -16,4 +16,9 @@ public interface AskDeviceDataService {
void createFolder(String nDid, String path);
void deleteFolder(String nDid, String path);
/**
* 实时数据请求报文
*/
void askRealData(String nDid, Integer idx, Integer size);
}

74
iot-access/access-boot/src/main/java/com/njcn/access/service/impl/AskDeviceDataServiceImpl.java
View File

@@ -7,6 +7,7 @@ import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.access.api.CsTopicFeignClient;
import com.njcn.access.enums.AccessEnum;
import com.njcn.access.enums.TypeEnum;
import com.njcn.access.pojo.dto.AskDataDto;
import com.njcn.access.pojo.dto.ControlDto;
import com.njcn.access.pojo.dto.ReqAndResDto;
import com.njcn.access.pojo.dto.file.FileRedisDto;
@@ -96,8 +97,8 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
ReqAndResDto.Req reqAndResParam = getAllPojo(mid,name);
publisher.send("/Pfm/DevFileCmd/V1/"+nDid,new Gson().toJson(reqAndResParam),1,false);
//这里使用简单的轮询，但建议考虑更高效的机制
for (int i = 0; i < 120; i++) {
Thread.sleep(2000);
for (int i = 0; i < 12; i++) {
Thread.sleep(10000);
Object object2 = redisUtil.getObjectByKey("downloadFilePath:"+name);
if (!Objects.isNull(object2)) {
result = true;
@@ -195,6 +196,39 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,mid);
}
@Override
public void askRealData(String nDid, Integer idx, Integer clDId) {
Object object = getDeviceMid(nDid);
if (!Objects.isNull(object)) {
mid = (Integer) object;
}
ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
reqAndResParam.setMid(mid);
reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_6.getCode()));
reqAndResParam.setExpire(-1);
//fixme 目前设备都是直连设备，因此did就是其本身，默认为1，后期涉及网关，此did是需要动态变化的
reqAndResParam.setDid(1);
AskDataDto askDataDto = new AskDataDto();
askDataDto.setCldid(clDId);
askDataDto.setDataAttr(1);
askDataDto.setDataType(4);
askDataDto.setOperate(1);
askDataDto.setStartTime(-1);
askDataDto.setEndTime(-1);
askDataDto.setRtDuration(30);
askDataDto.setDsNameIdx(idx);
reqAndResParam.setMsg(askDataDto);
log.info("askDevData的请求报文:" + new Gson().toJson(reqAndResParam));
publisher.send("/Pfm/DevCmd/V1/"+nDid, new Gson().toJson(reqAndResParam),1,false);
mid = mid + 1;
if (mid > 10000) {
mid = 1;
}
redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,mid);
}
public Object getDeviceMid(String nDid) {
return redisUtil.getObjectByKey(AppRedisKey.DEVICE_MID + nDid);
}
@@ -203,7 +237,6 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
return csTopicFeignClient.find(nDid).getData();
}
/**
* 全文件下载请求报文
*/
@@ -238,39 +271,4 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
reqAndResParam.setMsg(jsonObject);
return reqAndResParam;
}
/**
* 根据装置响应来判断是否询问下一帧数据
*/
public void sendNextStep(String fileName, String id, int mid,int step) {
try {
for (int i = 1; i < 4; i++) {
if (step == 0 ){
Thread.sleep(5000);
} else {
Thread.sleep(2000);
}
FileRedisDto fileRedisDto = (FileRedisDto) redisUtil.getObjectByKey(AppRedisKey.DOWNLOAD + fileName + mid);
if (Objects.isNull(fileRedisDto)) {
FileRedisDto failDto = new FileRedisDto();
failDto.setCode(400);
redisUtil.saveByKeyWithExpire(AppRedisKey.DOWNLOAD + fileName + mid,failDto,10L);
} else {
if (Objects.equals(fileRedisDto.getCode(),200)) {
break;
} else {
log.info("第" +i+"次尝试");
//尝试失败则设置code为400，如果装置响应了，则会将code置为200
FileRedisDto failDto = new FileRedisDto();
failDto.setCode(400);
redisUtil.saveByKeyWithExpire(AppRedisKey.DOWNLOAD + fileName + mid,failDto,10L);
ReqAndResDto.Req req = getPojo(mid,fileName,step);
publisher.send("/Pfm/DevFileCmd/V1/" + id, new Gson().toJson(req), 1, false);
}
}
}
} catch (Exception e) {
throw new BusinessException(AlgorithmResponseEnum.ASK_DEVICE_DIR_ERROR);
}
}
}

12
iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java
View File

@@ -9,6 +9,7 @@ import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.access.enums.AccessEnum;
import com.njcn.access.enums.AccessResponseEnum;
import com.njcn.access.enums.TypeEnum;
import com.njcn.access.mapper.OverlimitMapper;
import com.njcn.access.param.DevAccessParam;
import com.njcn.access.pojo.dto.AccessDto;
import com.njcn.access.pojo.dto.CsModelDto;
@@ -26,6 +27,8 @@ import com.njcn.csdevice.pojo.param.CsLedgerParam;
import com.njcn.csdevice.pojo.param.CsLineParam;
import com.njcn.csdevice.pojo.po.*;
import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
import com.njcn.device.biz.pojo.po.Overlimit;
import com.njcn.device.biz.utils.COverlimitUtil;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.api.DicDataFeignClient;
@@ -77,6 +80,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
private final ChannelObjectUtil channelObjectUtil;
private final CsLineFeignClient csLineFeignClient;
private final DataSetFeignClient dataSetFeignClient;
private final OverlimitMapper overlimitMapper;
@Override
@Transactional(rollbackFor = {Exception.class})
@@ -365,7 +369,6 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
return devAccessAskTemplate(nDid,version,1);
}
@Override
@Transactional(rollbackFor = {Exception.class})
public String wlDevRegister(String nDid) {
@@ -432,6 +435,13 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
csLedgerService.addLedgerTree(param);
});
csLineService.saveBatch(csLinePoList);
//生成监测点限值
for(CsLinePO csLinePO: csLinePoList){
Overlimit overlimit = COverlimitUtil.globalAssemble(csLinePO.getVolGrade().floatValue(),10f,10f,10f,0,1);
overlimit.setId(csLinePO.getLineId());
overlimitMapper.deleteById(csLinePO.getLineId());
overlimitMapper.insert(overlimit);
}
//4.生成装置和模板的关系表
CsDevModelRelationAddParm csDevModelRelationAddParm = new CsDevModelRelationAddParm();
csDevModelRelationAddParm.setDevId(vo.getId());

29
iot-analysis/analysis-rt/rt-api/pom.xml
View File

@@ -8,11 +8,36 @@
<version>1.0.0</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<name>rt-api</name>
<artifactId>rt-api</artifactId>
<version>1.0.0</version>
<name>rt-api</name>
<dependencies>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-db</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-microservice</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-mq</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

18
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/api/RtFeignClient.java Normal file
View File

@@ -0,0 +1,18 @@
package com.njcn.rt.api;
import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.rt.api.fallback.RtClientFallbackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
/**
* @author xy
*/
@FeignClient(value = ServerInfo.CS_RT_BOOT, path = "/rtData", fallbackFactory = RtClientFallbackFactory.class,contextId = "rtData")
public interface RtFeignClient {
@PostMapping("/rtAnalysis")
HttpResult<String> analysis(AppAutoDataMessage appAutoDataMessage);
}

35
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/api/fallback/RtClientFallbackFactory.java Normal file
View File

@@ -0,0 +1,35 @@
package com.njcn.rt.api.fallback;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.rt.api.RtFeignClient;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* @author xy
*/
@Slf4j
@Component
public class RtClientFallbackFactory implements FallbackFactory<RtFeignClient> {
@Override
public RtFeignClient create(Throwable cause) {
//判断抛出异常是否为解码器抛出的业务异常
Enum<?> exceptionEnum = CommonResponseEnum.SERVICE_FALLBACK;
if (cause.getCause() instanceof BusinessException) {
BusinessException businessException = (BusinessException) cause.getCause();
}
Enum<?> finalExceptionEnum = exceptionEnum;
return new RtFeignClient() {
@Override
public HttpResult<String> analysis(AppAutoDataMessage appAutoDataMessage) {
log.error("{}异常，降级处理，异常为：{}","实时数据解析",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

37
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/enums/RtResponseEnum.java Normal file
View File

@@ -0,0 +1,37 @@
package com.njcn.rt.enums;
import lombok.Getter;
/**
* @author xuyang
* @version 1.0.0
* @date 2023年04月17日 10:50
*/
@Getter
public enum RtResponseEnum {
/**
* A1001 ~ A1099 用于实时数据模块的枚举
* <p>
*/
RT_ERROR("A10001","实时数据模块错误"),
DATA_ARRAY_NULL("A10002","详细数据为空"),
AUTO_DATA_NULL("A10002","上送数据为空"),
DICT_NULL("A10002","字典数据为空"),
LINE_NULL("A10002","监测点为空"),
ARRAY_DATA_NOT_MATCH("A10003","上送数据与模板匹配失败"),
;
private final String code;
private final String message;
RtResponseEnum(String code, String message) {
this.code = code;
this.message = message;
}
}

200
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/pojo/dto/BaseRealDataSet.java Normal file
View File

@@ -0,0 +1,200 @@
package com.njcn.rt.pojo.dto;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
/**
* 实时数据-基础数据
*/
@Data
public class BaseRealDataSet implements Serializable {
@ApiModelProperty("监测点id")
private String lineId;
@ApiModelProperty("数据时间")
private String dataTime;
@ApiModelProperty("pt")
private Float pt;
@ApiModelProperty("ct")
private Float ct;
@ApiModelProperty("数据类型 Primary-一次值 Secondary-二次值")
private String dataLevel;
@ApiModelProperty("频率")
private Float freq;
@ApiModelProperty("频率偏差")
private Float freqDev;
@ApiModelProperty("A相-电压有效值")
private Float vRmsA;
@ApiModelProperty("B相-电压有效值")
private Float vRmsB;
@ApiModelProperty("C相-电压有效值")
private Float vRmsC;
// @ApiModelProperty("A相-相电压有效值")
// private Float vuRmsA;
//
// @ApiModelProperty("B相-相电压有效值")
// private Float vuRmsB;
//
// @ApiModelProperty("C相-相电压有效值")
// private Float vuRmsC;
//
// @ApiModelProperty("A相-线电压有效值")
// private Float vlRmsA;
//
// @ApiModelProperty("B相-线电压有效值")
// private Float vlRmsB;
//
// @ApiModelProperty("C相-线电压有效值")
// private Float vlRmsC;
@ApiModelProperty("A相-基波电压幅值")
private Float v1A;
@ApiModelProperty("B相-基波电压幅值")
private Float v1B;
@ApiModelProperty("C相-基波电压幅值")
private Float v1C;
@ApiModelProperty("A相-电流有效值")
private Float iRmsA;
@ApiModelProperty("B相-电流有效值")
private Float iRmsB;
@ApiModelProperty("C相-电流有效值")
private Float iRmsC;
@ApiModelProperty("A相-基波电流幅值")
private Float i1A;
@ApiModelProperty("B相-基波电流幅值")
private Float i1B;
@ApiModelProperty("C相-基波电流幅值")
private Float i1C;
@ApiModelProperty("A相-电压偏差")
private Float vDevA;
@ApiModelProperty("B相-电压偏差")
private Float vDevB;
@ApiModelProperty("C相-电压偏差")
private Float vDevC;
@ApiModelProperty("A相-基波电压相位")
private Float v1AngA;
@ApiModelProperty("B相-基波电压相位")
private Float v1AngB;
@ApiModelProperty("C相-基波电压相位")
private Float v1AngC;
@ApiModelProperty("A相-基波电流相位")
private Float i1AngA;
@ApiModelProperty("B相-基波电流相位")
private Float i1AngB;
@ApiModelProperty("C相-基波电流相位")
private Float i1AngC;
@ApiModelProperty("A相-电压总谐波畸变率")
private Float vThdA;
@ApiModelProperty("B相-电压总谐波畸变率")
private Float vThdB;
@ApiModelProperty("C相-电压总谐波畸变率")
private Float vThdC;
@ApiModelProperty("A相-电流总谐波畸变率")
private Float iThdA;
@ApiModelProperty("B相-电流总谐波畸变率")
private Float iThdB;
@ApiModelProperty("C相-电流总谐波畸变率")
private Float iThdC;
@ApiModelProperty("电压不平衡度")
private Float vUnbalance;
@ApiModelProperty("电流不平衡度")
private Float iUnbalance;
@ApiModelProperty("A相-有功功率")
private Float pA;
@ApiModelProperty("B相-有功功率")
private Float pB;
@ApiModelProperty("C相-有功功率")
private Float pC;
@ApiModelProperty("A相-无功功率")
private Float qA;
@ApiModelProperty("B相-无功功率")
private Float qB;
@ApiModelProperty("C相-无功功率")
private Float qC;
@ApiModelProperty("A相-视在功率")
private Float sA;
@ApiModelProperty("B相-视在功率")
private Float sB;
@ApiModelProperty("C相-视在功率")
private Float sC;
@ApiModelProperty("A相-功率因数")
private Float pfA;
@ApiModelProperty("B相-功率因数")
private Float pfB;
@ApiModelProperty("C相-功率因数")
private Float pfC;
@ApiModelProperty("A相-基波功率因数")
private Float dpfA;
@ApiModelProperty("B相-基波功率因数")
private Float dpfB;
@ApiModelProperty("C相-基波功率因数")
private Float dpfC;
@ApiModelProperty("总-有功功率")
private Float pTot;
@ApiModelProperty("总-无功功率")
private Float qTot;
@ApiModelProperty("总-视在功率")
private Float sTot;
@ApiModelProperty("总-功率因数")
private Float pfTot;
@ApiModelProperty("总-基波功率因数")
private Float dpfTot;
}

25
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/pojo/dto/HarmData.java Normal file
View File

@@ -0,0 +1,25 @@
package com.njcn.rt.pojo.dto;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
/**
* @author xy
*/
@Data
public class HarmData implements Serializable {
@ApiModelProperty("指标名称")
private String harmName;
@ApiModelProperty("相别")
private String phase;
@ApiModelProperty("数据")
private Float data;
@ApiModelProperty("排序")
private Integer sort;
}

81
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/pojo/dto/HarmRealDataSet.java Normal file
View File

@@ -0,0 +1,81 @@
package com.njcn.rt.pojo.dto;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
/**
* 实时数据-谐波数据
* @author xy
*/
@Data
public class HarmRealDataSet implements Serializable {
@ApiModelProperty("监测点id")
private String lineId;
@ApiModelProperty("数据时间")
private String dataTime;
@ApiModelProperty("pt")
private Float pt;
@ApiModelProperty("ct")
private Float ct;
@ApiModelProperty("数据类型 Primary-一次值 Secondary-二次值")
private String dataLevel;
private Float data1;
private Float data2;
private Float data3;
private Float data4;
private Float data5;
private Float data6;
private Float data7;
private Float data8;
private Float data9;
private Float data10;
private Float data11;
private Float data12;
private Float data13;
private Float data14;
private Float data15;
private Float data16;
private Float data17;
private Float data18;
private Float data19;
private Float data20;
private Float data21;
private Float data22;
private Float data23;
private Float data24;
private Float data25;
private Float data26;
private Float data27;
private Float data28;
private Float data29;
private Float data30;
private Float data31;
private Float data32;
private Float data33;
private Float data34;
private Float data35;
private Float data36;
private Float data37;
private Float data38;
private Float data39;
private Float data40;
private Float data41;
private Float data42;
private Float data43;
private Float data44;
private Float data45;
private Float data46;
private Float data47;
private Float data48;
private Float data49;
private Float data50;
}

21
iot-analysis/analysis-rt/rt-boot/pom.xml
View File

@@ -40,6 +40,27 @@
<artifactId>common-db</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-mq</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>rt-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>cs-device-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>access-api</artifactId>
<version>1.0.0</version>
<scope>compile</scope>
</dependency>
</dependencies>
<build>

2
iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/RtBootApplication.java
View File

@@ -5,6 +5,7 @@ import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.DependsOn;
/**
@@ -13,6 +14,7 @@ import org.springframework.cloud.openfeign.EnableFeignClients;
* @date 2021年12月09日 20:59
*/
@Slf4j
@DependsOn("proxyMapperRegister")
@MapperScan("com.njcn.**.mapper")
@EnableFeignClients(basePackages = "com.njcn")
@SpringBootApplication(scanBasePackages = "com.njcn")

23
iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/controller/RtController.java
View File

@@ -5,6 +5,8 @@ import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.rt.service.IRtService;
import com.njcn.web.controller.BaseController;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
@@ -31,16 +33,17 @@ import org.springframework.web.bind.annotation.RestController;
@AllArgsConstructor
public class RtController extends BaseController {
// @OperateInfo(info = LogEnum.BUSINESS_COMMON)
// @PostMapping("/analysis")
// @ApiOperation("数据解析")
// @ApiImplicitParam(name = "csDataEffectiveAddParm", value = "新增app数据有效性表参数", required = true)
// public HttpResult<Boolean> addDataEffective(@RequestBody @Validated CsDataEffectiveAddParm csDataEffectiveAddParm){
// String methodDescribe = getMethodDescribe("addDataEffective");
//
// boolean save = csDataEffectiveService.add (csDataEffectiveAddParm);
// return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, save, methodDescribe);
// }
private final IRtService rtService;
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/rtAnalysis")
@ApiOperation("实时数据解析")
@ApiImplicitParam(name = "appAutoDataMessage", value = "数据实体", required = true)
public HttpResult<String> analysis(@RequestBody @Validated AppAutoDataMessage appAutoDataMessage){
String methodDescribe = getMethodDescribe("analysis");
rtService.analysis(appAutoDataMessage);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
}

11
iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/IRtService.java Normal file
View File

@@ -0,0 +1,11 @@
package com.njcn.rt.service;
import com.njcn.mq.message.AppAutoDataMessage;
/**
* @author xy
*/
public interface IRtService {
void analysis(AppAutoDataMessage appAutoDataMessage);
}

275
iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java Normal file
View File

@@ -0,0 +1,275 @@
package com.njcn.rt.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.DataArrayFeignClient;
import com.njcn.csdevice.api.DataSetFeignClient;
import com.njcn.csdevice.pojo.po.CsDataArray;
import com.njcn.csdevice.pojo.po.CsDataSet;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.rt.enums.RtResponseEnum;
import com.njcn.rt.pojo.dto.BaseRealDataSet;
import com.njcn.rt.pojo.dto.HarmData;
import com.njcn.rt.pojo.dto.HarmRealDataSet;
import com.njcn.rt.service.IRtService;
import com.njcn.web.utils.FloatUtils;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import java.lang.reflect.Field;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* @author xy
*/
@Service
@RequiredArgsConstructor
public class RtServiceImpl implements IRtService {
private final CsLineFeignClient csLineFeignClient;
private final DataSetFeignClient dataSetFeignClient;
private final DataArrayFeignClient dataArrayFeignClient;
private final RedisUtil redisUtil;
private final ChannelObjectUtil channelObjectUtil;
private final MqttPublisher publisher;
@Override
public void analysis(AppAutoDataMessage appAutoDataMessage) {
List<CsDataArray> dataArrayList;
//监测点id
String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
//获取监测点基础信息
CsLinePO po = csLineFeignClient.getById(lineId).getData();
//获取数据集 dataSet
Integer idx = appAutoDataMessage.getMsg().getDsNameIdx();
CsDataSet dataSet = dataSetFeignClient.getDataSetByIdx(po.getDataModelId(),idx).getData();
//根据数据集获取指标 dataArray
//实时数据数据集不区分最大最小类型，因此数据集取平均值用于解析
String key = "BaseRealData:" + lineId + idx;
Object object = redisUtil.getObjectByKey(key);
if (Objects.isNull(object)){
dataArrayList = saveBaseRealDataSet(key,dataSet.getId());
} else {
dataArrayList = channelObjectUtil.objectToList(object,CsDataArray.class);
}
//根据dataArray解析数据
AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
//fixme 这边先根据数据集的名称来返回对应实体，这边感觉不太合适，后期有好方案再调整
//基础数据
if (Objects.equals(dataSet.getName(),"Ds$Pqd$Rt$Basic$01")) {
BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType());
baseRealDataSet.setLineId(lineId);
baseRealDataSet.setPt(po.getPtRatio().floatValue());
baseRealDataSet.setCt(po.getCtRatio().floatValue());
baseRealDataSet.setDataLevel(dataSet.getDataLevel());
long timestamp = item.getDataTimeSec() - 8*3600;
baseRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
}
//fixme 目前实时数据只有基础数据和谐波数据，后期拓展，这边需要再判断
else {
HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item);
harmRealDataSet.setLineId(lineId);
harmRealDataSet.setPt(po.getPtRatio().floatValue());
harmRealDataSet.setCt(po.getCtRatio().floatValue());
harmRealDataSet.setDataLevel(dataSet.getDataLevel());
long timestamp = item.getDataTimeSec() - 8*3600;
harmRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false);
}
}
/**
* 时间处理
*/
public String getTime(long timestamp) {
Instant instant = Instant.ofEpochSecond(timestamp);
LocalDateTime dateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
return dateTime.format(formatter);
}
/**
* 缓存实时数据数据集
* @param key
* @param dataSetId
* @return
*/
public List<CsDataArray> saveBaseRealDataSet(String key, String dataSetId) {
List<CsDataArray> dataArrays = dataArrayFeignClient.getArrayBySet(dataSetId).getData();
List<CsDataArray> dataArrayList = dataArrays.stream().filter(item->Objects.equals(item.getStatMethod(),"avg")).collect(Collectors.toList());
redisUtil.saveByKeyWithExpire(key,dataArrayList,600L);
return dataArrayList;
}
/**
* 数据解码
* @return
*/
public Map<String,Float> getData(List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray dataArray) {
Map<String,Float> dataMap = new LinkedHashMap<>();
//解码
List<Float> floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(dataArray.getData()));
if (CollectionUtil.isEmpty(floats)){
throw new BusinessException(RtResponseEnum.AUTO_DATA_NULL);
}
//校验模板和解码数据数量能否对应上
if (!Objects.equals(dataArrayList.size(),floats.size())){
throw new BusinessException(RtResponseEnum.ARRAY_DATA_NOT_MATCH);
}
for (int i = 0; i < dataArrayList.size(); i++) {
dataMap.put(dataArrayList.get(i).getName() + dataArrayList.get(i).getPhase(),floats.get(i));
}
return dataMap;
}
public BaseRealDataSet assembleData(List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray dataArray,Integer conType) {
Map<String,Float> dataMap = getData(dataArrayList,dataArray);
return channelData(dataMap,conType);
}
public BaseRealDataSet channelData(Map<String,Float> map,Integer conType) {
BaseRealDataSet baseRealDataSet = new BaseRealDataSet();
//频率
baseRealDataSet.setFreq(FloatUtils.get2Float(map.get("Pq_FreqM")));
//频率偏差
baseRealDataSet.setFreqDev(FloatUtils.get2Float(map.get("Pq_FreqDevM")));
//判断监测点的接线方式，不同接线方式电压有效值取值不同
//星型-相电压 角形、V型-线电压
//电压有效值
if (conType == 0) {
baseRealDataSet.setVRmsA(FloatUtils.get2Float(map.get("Pq_RmsUA")));
baseRealDataSet.setVRmsB(FloatUtils.get2Float(map.get("Pq_RmsUB")));
baseRealDataSet.setVRmsC(FloatUtils.get2Float(map.get("Pq_RmsUC")));
} else {
baseRealDataSet.setVRmsA(FloatUtils.get2Float(map.get("Pq_RmsLUAB")));
baseRealDataSet.setVRmsB(FloatUtils.get2Float(map.get("Pq_RmsLUBC")));
baseRealDataSet.setVRmsC(FloatUtils.get2Float(map.get("Pq_RmsLUCA")));
}
//基波电压幅值
baseRealDataSet.setV1A(FloatUtils.get2Float(map.get("Pq_RmsFundUA")));
baseRealDataSet.setV1B(FloatUtils.get2Float(map.get("Pq_RmsFundUB")));
baseRealDataSet.setV1C(FloatUtils.get2Float(map.get("Pq_RmsFundUC")));
//电流有效值
baseRealDataSet.setIRmsA(FloatUtils.get2Float(map.get("Pq_RmsIA")));
baseRealDataSet.setIRmsB(FloatUtils.get2Float(map.get("Pq_RmsIB")));
baseRealDataSet.setIRmsC(FloatUtils.get2Float(map.get("Pq_RmsIC")));
//基波电流幅值
baseRealDataSet.setI1A(FloatUtils.get2Float(map.get("Pq_RmsFundIA")));
baseRealDataSet.setI1B(FloatUtils.get2Float(map.get("Pq_RmsFundIB")));
baseRealDataSet.setI1C(FloatUtils.get2Float(map.get("Pq_RmsFundIC")));
//电压偏差
baseRealDataSet.setVDevA(FloatUtils.get2Float(map.get("Pq_UDevA")));
baseRealDataSet.setVDevB(FloatUtils.get2Float(map.get("Pq_UDevB")));
baseRealDataSet.setVDevC(FloatUtils.get2Float(map.get("Pq_UDevC")));
//基波电压相位
baseRealDataSet.setV1AngA(FloatUtils.get2Float(map.get("Pq_FundUAngA")));
baseRealDataSet.setV1AngB(FloatUtils.get2Float(map.get("Pq_FundUAngB")));
baseRealDataSet.setV1AngC(FloatUtils.get2Float(map.get("Pq_FundUAngC")));
//基波电流相位
baseRealDataSet.setI1AngA(FloatUtils.get2Float(map.get("Pq_FundIAngA")));
baseRealDataSet.setI1AngB(FloatUtils.get2Float(map.get("Pq_FundIAngB")));
baseRealDataSet.setI1AngC(FloatUtils.get2Float(map.get("Pq_FundIAngC")));
//电压总谐波畸变率
baseRealDataSet.setVThdA(FloatUtils.get2Float(map.get("Pq_ThdUA")));
baseRealDataSet.setVThdB(FloatUtils.get2Float(map.get("Pq_ThdUB")));
baseRealDataSet.setVThdC(FloatUtils.get2Float(map.get("Pq_ThdUC")));
//电流总谐波畸变率
baseRealDataSet.setIThdA(FloatUtils.get2Float(map.get("Pq_ThdIA")));
baseRealDataSet.setIThdB(FloatUtils.get2Float(map.get("Pq_ThdIB")));
baseRealDataSet.setIThdC(FloatUtils.get2Float(map.get("Pq_ThdIC")));
//电压不平衡度
baseRealDataSet.setVUnbalance(FloatUtils.get2Float(map.get("Pq_UnbalNegUM")));
//电流不平衡度
baseRealDataSet.setIUnbalance(FloatUtils.get2Float(map.get("Pq_UnbalNegIM")));
//有功功率
baseRealDataSet.setPA(FloatUtils.get2Float(map.get("Pq_PA")));
baseRealDataSet.setPB(FloatUtils.get2Float(map.get("Pq_PB")));
baseRealDataSet.setPC(FloatUtils.get2Float(map.get("Pq_PC")));
baseRealDataSet.setPTot(FloatUtils.get2Float(map.get("Pq_TotPM")));
//无功功率
baseRealDataSet.setQA(FloatUtils.get2Float(map.get("Pq_QA")));
baseRealDataSet.setQB(FloatUtils.get2Float(map.get("Pq_QB")));
baseRealDataSet.setQC(FloatUtils.get2Float(map.get("Pq_QC")));
baseRealDataSet.setQTot(FloatUtils.get2Float(map.get("Pq_TotQM")));
//视在功率
baseRealDataSet.setSA(FloatUtils.get2Float(map.get("Pq_SA")));
baseRealDataSet.setSB(FloatUtils.get2Float(map.get("Pq_SB")));
baseRealDataSet.setSC(FloatUtils.get2Float(map.get("Pq_SC")));
baseRealDataSet.setSTot(FloatUtils.get2Float(map.get("Pq_TotSM")));
//功率因数
baseRealDataSet.setPfA(FloatUtils.get2Float(map.get("Pq_PFA")));
baseRealDataSet.setPfB(FloatUtils.get2Float(map.get("Pq_PFB")));
baseRealDataSet.setPfC(FloatUtils.get2Float(map.get("Pq_PFC")));
baseRealDataSet.setPfTot(FloatUtils.get2Float(map.get("Pq_TotPFM")));
//基波功率因数
baseRealDataSet.setDpfA(FloatUtils.get2Float(map.get("Pq_DPFA")));
baseRealDataSet.setDpfB(FloatUtils.get2Float(map.get("Pq_DPFB")));
baseRealDataSet.setDpfC(FloatUtils.get2Float(map.get("Pq_DPFC")));
baseRealDataSet.setDpfTot(FloatUtils.get2Float(map.get("Pq_TotDPFM")));
return baseRealDataSet;
}
public HarmRealDataSet harmData(List<CsDataArray> dataArrayList, AppAutoDataMessage.DataArray dataArray) {
HarmRealDataSet harmRealDataSet = new HarmRealDataSet();
List<HarmData> harmDataList = new ArrayList<>();
//解码
List<Float> floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(dataArray.getData()));
if (CollectionUtil.isEmpty(floats)){
throw new BusinessException(RtResponseEnum.AUTO_DATA_NULL);
}
//校验模板和解码数据数量能否对应上
if (!Objects.equals(dataArrayList.size(),floats.size())){
throw new BusinessException(RtResponseEnum.ARRAY_DATA_NOT_MATCH);
}
for (int i = 0; i < dataArrayList.size(); i++) {
HarmData harmData = new HarmData();
harmData.setHarmName(dataArrayList.get(i).getName());
harmData.setPhase(dataArrayList.get(i).getPhase());
harmData.setSort(dataArrayList.get(i).getSort());
harmData.setData(floats.get(i));
harmDataList.add(harmData);
}
//根据名称分组，然后在不同相别的数据中取最大值
List<HarmData> maxDataList = new ArrayList<>(harmDataList.stream()
.collect(Collectors.toMap(
HarmData::getHarmName,
Function.identity(),
BinaryOperator.maxBy(Comparator.comparingDouble(HarmData::getData))
))
.values());
//通过反射将数据赋值
Class<?> clazz = HarmRealDataSet.class;
maxDataList.forEach(item->{
if (Objects.equals(item.getHarmName(),"Pq_RmsFundI") || Objects.equals(item.getHarmName(),"Pq_RmsFundU")) {
harmRealDataSet.setData1(FloatUtils.get2Float(item.getData()));
} else {
String numberStr = item.getHarmName().substring(item.getHarmName().lastIndexOf('_') + 1);
String fieldName = "data" + numberStr;
try {
Field field = clazz.getDeclaredField(fieldName);
field.setAccessible(true);
field.set(harmRealDataSet,FloatUtils.get2Float(item.getData()));
} catch (Exception e) {
throw new RuntimeException(e);
}
}
});
return harmRealDataSet;
}
}

12
iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/IWlRecordService.java
View File

@@ -1,12 +0,0 @@
//package com.njcn.stat.service;
//
//import com.njcn.mq.message.AppAutoDataMessage;
//
///**
// * @author xy
// */
//public interface IWlRecordService {
//
// void addOrUpdateBaseData(AppAutoDataMessage appAutoDataMessage);
//
//}

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

@@ -11,8 +11,6 @@ import com.njcn.csdevice.pojo.param.DataArrayParam;
import com.njcn.csdevice.pojo.po.CsDataArray;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
import com.njcn.cswarn.api.CsEquipmentAlarmFeignClient;
import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
import com.njcn.influx.utils.InfluxDbUtils;
import com.njcn.mq.message.AppAutoDataMessage;
@@ -24,7 +22,6 @@ import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.api.DictTreeFeignClient;
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 lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

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

@@ -1,44 +0,0 @@
//package com.njcn.stat.service.impl;
//
//import com.njcn.csdevice.api.EquipmentFeignClient;
//import com.njcn.csdevice.api.WlRecordFeignClient;
//import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
//import com.njcn.csdevice.pojo.po.WlRecord;
//import com.njcn.mq.message.AppAutoDataMessage;
//import com.njcn.stat.service.IWlRecordService;
//import lombok.AllArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.BeanUtils;
//import org.springframework.stereotype.Service;
//
///**
// * 类的介绍：
// *
// * @author xuyang
// * @version 1.0.0
// * @createTime 2023/8/14 9:32
// */
//@Service
//@Slf4j
//@AllArgsConstructor
//public class WlRecordServiceImpl implements IWlRecordService{
//
// private final EquipmentFeignClient equipmentFeignClient;
//
// private final WlRecordFeignClient wlRecordFeignClient;
//
// @Override
// public void addOrUpdateBaseData(AppAutoDataMessage appAutoDataMessage) {
// WlRecord wlRecord = new WlRecord();
// CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(appAutoDataMessage.getId()).getData();
// AppAutoDataMessage.DataArray dataArray = appAutoDataMessage.getMsg().getDataArray().get(0);
// BeanUtils.copyProperties(dataArray, wlRecord);
// wlRecord.setDevId(po.getId());
// wlRecord.setLineId(po.getNdid() + appAutoDataMessage.getMsg().getClDid());
// wlRecord.setGcDataPath(dataArray.getPrjDataPath());
// if (dataArray.getPrjTimeEnd() == -1) {
// wlRecord.setEndTime(null);
// }
// wlRecordFeignClient.addBaseData(wlRecord);
// }
//}

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

@@ -35,7 +35,6 @@ 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.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

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

@@ -131,12 +131,8 @@ public class FileServiceImpl implements IFileService {
String lsFileName = generalInfo.getBusinessTempPath() + File.separator + fileName.split(StrUtil.SLASH)[fileName.split(StrUtil.SLASH).length - 1];
File lsFile = new File(generalInfo.getBusinessTempPath());
//如果文件夹不存在则创建
if (!lsFile.exists()) {
if (!lsFile.isDirectory()) {
if (!lsFile.exists() && !lsFile.isDirectory()) {
lsFile .mkdirs();
} else {
lsFile.createNewFile();
}
}
//获取缓存的文件信息
Object fileInfo = redisUtil.getObjectByKey(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(fileName));

5
iot-message/message-boot/pom.xml
View File

@@ -61,6 +61,11 @@
<artifactId>stat-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>rt-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>system-api</artifactId>

8
iot-message/message-boot/src/main/java/com/njcn/message/consumer/AppAutoDataConsumer.java
View File

@@ -8,6 +8,7 @@ import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.rt.api.RtFeignClient;
import com.njcn.stat.api.StatFeignClient;
import com.njcn.system.api.RocketMqLogFeignClient;
import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
@@ -39,14 +40,14 @@ public class AppAutoDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDa
@Resource
private StatFeignClient statFeignClient;
@Resource
private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Resource
private RtFeignClient rtFeignClient;
// @Resource
// private WlRecordFeignClient wlRecordFeignClient;
// private WlRecordFeignClient wlRecordFeignClient
@Override
protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
@@ -57,6 +58,7 @@ public class AppAutoDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDa
break;
case 1:
log.info("分发至实时数据");
rtFeignClient.analysis(appAutoDataMessage);
break;
case 2:
log.info(appAutoDataMessage.getKey() + "分发至统计数据");