Schedule定时任务。定时执行，将结果放在指定的表中。LineiceSchedule.java 判段装置接受的数据是否为空，过多，过少;判定(EQUAL_ICETHICKNESS )等值覆冰厚度;检查超时接受的数据。MonitorDataSchedule.java 筛选同一id对应的所有IP，和同一IP对应所有id。
方法名上加注解@Scheduled，cron表达式 控制执行时间与频率。例如 @Scheduled(cron = "0 0 2 * * ?")
启动类上加注解@EnableScheduling。
LineiceSchedule.java
package com.evan.schedule;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TimeZone;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import com.evan.bean.SdCagErrormonitors;
import com.evan.bean.SdCagLineice;
import com.evan.bean.SdCagParam;
import com.evan.constant.Constant;
import com.evan.service.ISdCagErrordatasService;
import com.evan.service.ISdCagErrormonitorsService;
import com.evan.service.ISdCagLineiceService;
import com.evan.service.ISdCagParamService;
import com.evan.service.ITsDeviceService;
@Service
public class LineiceSchedule{
@Autowired
private ISdCagErrormonitorsService sdCagErrormonitorsService;
@Autowired
private ISdCagLineiceService sdCagLineiceService;
@Autowired
private ITsDeviceService tsDeviceService;
@Autowired
private ISdCagParamService sdCagParamService;
@Autowired
private ISdCagErrordatasService sdCagErrordatasService;
/**
* 判段装置接受的数据是否为空，过多，过少
*/
@Scheduled(cron = "0 0 2 * * ?")
public void analyse() {
long end = System.currentTimeMillis()/(1000*3600*24)*(1000*3600*24)-TimeZone.getDefault().getRawOffset();
long start = end - 24*60*60*1000;
String type = Constant.MTYPE_LINEICE;
SdCagErrormonitors errormonitors = null;
List<SdCagErrormonitors> errormonitorsList = null;
List<Map<String, Object>> countList = sdCagLineiceService.getCount(start, end);
List<String> idList = tsDeviceService.getMonitorsByMtype(type);
//查询前一天错误装置cmdid
List<String> cmdidErrordataseList = sdCagErrordatasService.getSdCagErrordataseList(start, end);
//先去除错误装置
idList.removeAll(cmdidErrordataseList);
SdCagParam sdCagParamLineiceMin = sdCagParamService.getSdCagParam(Constant.PARAM_LINEICE_MIN);
SdCagParam sdCagParamLineiceMax = sdCagParamService.getSdCagParam(Constant.PARAM_LINEICE_MAX);
float lineiceMin = sdCagParamLineiceMin.getValue();
float lineiceMax = sdCagParamLineiceMax.getValue();
errormonitorsList = new ArrayList<SdCagErrormonitors>();
for (String id : idList) {
int quantity = 0;
for (Map<String, Object> count : countList) {
if (id.equals(count.get("cmdId"))) {
quantity = ((Long)count.get("quantity")).intValue();
break;
}
}
if (quantity == 0) {
errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(id);
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_EMPTY);
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
errormonitorsList.add(errormonitors);
} else if (quantity >= lineiceMax) {
errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(id);
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_CNT_TOOHIGH);
errormonitors.setError_msg("当前数据量:" + quantity + ";最大数据量:" + lineiceMax);
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
errormonitorsList.add(errormonitors);
} else if (quantity > 0 && quantity < lineiceMin) {
errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(id);
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_CNT_TOOLOW);
errormonitors.setError_msg("当前数据量:" + quantity + ";最小数据量:" + lineiceMin);
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
errormonitorsList.add(errormonitors);
}
}
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_EMPTY, start, end);
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_CNT_TOOHIGH, start, end);
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_CNT_TOOLOW, start, end);
boolean state = sdCagErrormonitorsService.batchAddSdCagErrormonitors(errormonitorsList);
System.out.println("Lineice----" + state);
}
/**
* 判定 EQUAL_ICETHICKNESS 等值覆冰厚度
*/
@Scheduled(cron = "0 0 2 * * ?")
public void analyseIcethickness() {
long end = System.currentTimeMillis()/(1000*3600*24)*(1000*3600*24)-TimeZone.getDefault().getRawOffset();
long start = end - 24*60*60*1000;
String type = Constant.MTYPE_LINEICE;
List<String> idList = tsDeviceService.getMonitorsByMtype(type);
//查询前一天错误装置cmdid
List<String> cmdidErrordataseList = sdCagErrordatasService.getSdCagErrordataseList(start, end);
idList.removeAll(cmdidErrordataseList);
List<SdCagLineice> lineiceList = sdCagLineiceService.getSdCagLineiceList(start, end);
float icethicknessMin = (sdCagParamService.getSdCagParam(Constant.PARAM_ICETHICKNESS_MIN)).getValue();
float icethicknessMax = (sdCagParamService.getSdCagParam(Constant.PARAM_ICETHICKNESS_MAX)).getValue();
List<SdCagErrormonitors> errormonitorsList = new ArrayList<SdCagErrormonitors>();
Map<String, Set<Double>> icethicknessMap = new HashMap<String,Set<Double>>();
for(String id : idList) {
Set<Double> icethicknessSet = new HashSet<Double>();
for(SdCagLineice lineice : lineiceList) {
if(id.equals(lineice.getCmd_id())) {
icethicknessSet.add(lineice.getEqual_icethickness());
}
}
icethicknessMap.put(id, icethicknessSet);
}
for(String id : idList) {
for(SdCagLineice lineice : lineiceList) {
if(id.equals(lineice.getCmd_id())) {
if(lineice.getEqual_icethickness() > icethicknessMax) {
SdCagErrormonitors errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(id);
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_DATA_TOOHIGH);
errormonitors.setError_msg("等值覆冰厚度过高，数据值为:" + lineice.getEqual_icethickness());
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
errormonitorsList.add(errormonitors);
}
if(lineice.getEqual_icethickness() < icethicknessMin) {
SdCagErrormonitors errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(id);
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_DATA_TOOLOW);
errormonitors.setError_msg("等值覆冰厚度过低，数据值为:" + lineice.getEqual_icethickness());
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
errormonitorsList.add(errormonitors);
}
}
}
}
for(Map.Entry<String, Set<Double>> entry : icethicknessMap.entrySet()) {
List<Double> icethicknessList = new ArrayList<Double>(entry.getValue());
if(entry.getValue() != null && entry.getValue().size() == 1 && icethicknessList.get(0) != 0.0) {
SdCagErrormonitors errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(entry.getKey());
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_UNCHANGED);
errormonitors.setError_msg("等值覆冰厚度恒定为:" + icethicknessList.get(0));
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
errormonitorsList.add(errormonitors);
}
}
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_UNCHANGED, start, end);
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_DATA_TOOHIGH, start, end);
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_DATA_TOOLOW, start, end);
boolean state = sdCagErrormonitorsService.batchAddSdCagErrormonitors(errormonitorsList);
System.out.println("analyseIcethickness----" + state);
}
/**
* 检查超时接受的数据
**/
@Scheduled(cron = "0 0 2 * * ?")
public void checkRecvtime() {
long end = System.currentTimeMillis()/(1000*3600*24)*(1000*3600*24)-TimeZone.getDefault().getRawOffset();
long start = end - 24*60*60*1000;
String type = Constant.MTYPE_LINEICE;
//查询前一天错误装置cmdid
List<String> cmdidErrordataseList = sdCagErrordatasService.getSdCagErrordataseList(start, end);
SdCagErrormonitors errormonitors = null;
List<SdCagLineice> lineiceList = sdCagLineiceService.getSdCagLineiceList(start, end);
SdCagParam recvtimeInvalid = sdCagParamService.getSdCagParam(Constant.PARAM_RECVTIME_INVALID);
float recvtimeInvalidVal = recvtimeInvalid.getValue();
SdCagParam frequency = sdCagParamService.getSdCagParam(Constant.PARAM_RECVTIME_FREQUENCY);
int frequencyVal = (int) frequency.getValue();
Map<String, Integer> lineiceMap = new HashMap<String, Integer>();
int i = 1;
for(SdCagLineice lineice :lineiceList) {
float timeChange = (lineice.getRecvtime()-lineice.getTime_stamp())/1000/60;
if(!cmdidErrordataseList.contains(lineice.getCmd_id()) && timeChange > recvtimeInvalidVal) {
if (lineiceMap.containsKey(lineice.getCmd_id())) {
Integer idNum = lineiceMap.get(lineice.getCmd_id());
lineiceMap.put(lineice.getCmd_id(), ++idNum);
} else {
lineiceMap.put(lineice.getCmd_id(), i);
}
}
}
List<SdCagErrormonitors> recvtimeList = new ArrayList<SdCagErrormonitors>();
for (Entry<String, Integer> entry : lineiceMap.entrySet()) {
if (entry.getValue() > frequencyVal) {
errormonitors = new SdCagErrormonitors();
errormonitors.setCmd_id(entry.getKey());
errormonitors.setType(type);
errormonitors.setError_type(Constant.ERROR_TYPE_RECVTIME_INVALID);
errormonitors.setError_msg("当前数据量:"+entry.getValue()+"条;容错量:"+frequencyVal + "条");
errormonitors.setTime(start);
errormonitors.setTime_stamp(start);
recvtimeList.add(errormonitors);
}
}
sdCagErrormonitorsService.deleteSdCagErrormonitors(type, Constant.ERROR_TYPE_RECVTIME_INVALID, start, end);
boolean state = sdCagErrormonitorsService.batchAddSdCagErrormonitors(recvtimeList);
System.out.println("Lineice--checkRecvtime" + state);
}
}
MonitorDataSchedule.java
package com.evan.schedule;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.TimeZone;
import java.util.stream.Collectors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import com.evan.bean.SdCagErrordatas;
import com.evan.bean.SdCagFromipId;
import com.evan.bean.VMonitorData;
import com.evan.constant.Constant;
import com.evan.service.ISdCagErrordatasService;
import com.evan.service.ISdCagFromipIdService;
@Service
public class MonitorDataSchedule{
@Autowired
private ISdCagErrordatasService sdCagErrordatasService;
@Autowired
private ISdCagFromipIdService sdCagFromipIdservice;
@Scheduled(cron = "0 30 2 * * ?")
public void analyseMonitorData() {
long begin = System.currentTimeMillis();
long end = System.currentTimeMillis()/(1000*3600*24)*(1000*3600*24)-TimeZone.getDefault().getRawOffset();
long start = end - 24*60*60*1000;
String sameId = Constant.ERROR_IP_SAMEID;
String sameIp = Constant.ERROR_IP_SAMEIP;
SdCagErrordatas sdCagErrordatas = null;
List<SdCagErrordatas> sdCagErrordatasList= null;
//防止重复执行产生重复数据
sdCagErrordatasService.batchDeleteByTime(start, end);
//得到的List中 只包含distinct 后 cmd_id fromip 字段
List<VMonitorData> vMonitorDataList = sdCagErrordatasService.getMonitorDataList(start, end);
System.out.println(vMonitorDataList.size());
List<String> cmdIdList = new ArrayList<String>();
List<String> fromipList = new ArrayList<String>();
for(VMonitorData vMonitorData : vMonitorDataList) {
cmdIdList.add(vMonitorData.getCmd_id());
fromipList.add(vMonitorData.getFromip() );
}
//获取装置表fromipid的所有信息
List<SdCagFromipId> sdCagFromipIdList = sdCagFromipIdservice.getSdCagFromipIdList();
//将id数大于1的筛选出来
List<String> sameIdList = getDuplicateElements(cmdIdList);
sdCagErrordatasList = new ArrayList<SdCagErrordatas>();
for(String id : sameIdList) {
List<String> ipList = new ArrayList<String>();
//取同id对应的ip
for(VMonitorData vMonitorData :vMonitorDataList) {
if(id.equals(vMonitorData.getCmd_id())){
ipList.add(vMonitorData.getFromip() );
}
}
for(String ip:ipList) {
sdCagErrordatas = new SdCagErrordatas();
sdCagErrordatas.setCmd_id(id);
sdCagErrordatas.setFromip(ip);
sdCagErrordatas.setType(sameId);
sdCagErrordatas.setError_msg("同一ID:"+id+",对应多个IP"+ipList.toString());
sdCagErrordatas.setTime_stamp(start);
sdCagErrordatas.setRecvtime(start);
sdCagErrordatas.setCreatime(start);
sdCagErrordatasList.add(sdCagErrordatas);
}
}
sdCagErrordatasService.batchAddSdCagErrordatas(sdCagErrordatasList);
//将fromip数大于1的筛选出来
List<String> sameIpList = getDuplicateElements(fromipList);
sdCagErrordatasList = new ArrayList<SdCagErrordatas>();
//获取同fromip对应的id
for(String ip : sameIpList) {
Set<String> poleSet = new HashSet<String>();
List<String> idList = new ArrayList<String>();
for(VMonitorData vMonitorData :vMonitorDataList) {
if(ip.equals(vMonitorData.getFromip())) {
idList.add(vMonitorData.getCmd_id());
poleSet.add(vMonitorData.getPole_id());
}
}
if(poleSet.size() > 1 ) {
List<String> cmdIdListFromipId = new ArrayList<String>();
for(SdCagFromipId sdCagFromipId : sdCagFromipIdList) {
if(ip.equals(sdCagFromipId.getFromip())){
cmdIdListFromipId.add(sdCagFromipId.getCmd_id());
}
}
//求出差集
idList.removeAll(cmdIdListFromipId);
for(String id:idList) {
sdCagErrordatas = new SdCagErrordatas();
sdCagErrordatas.setCmd_id(id);
sdCagErrordatas.setFromip(ip);
sdCagErrordatas.setType(sameIp);
sdCagErrordatas.setError_msg("同一IP:"+ip+",对应多个ID"+idList.toString());
sdCagErrordatas.setTime_stamp(start);
sdCagErrordatas.setRecvtime(start);
sdCagErrordatas.setCreatime(start);
sdCagErrordatasList.add(sdCagErrordatas);
}
}
}
sdCagErrordatasService.batchAddSdCagErrordatas(sdCagErrordatasList);
System.out.println("analyseMonitorData 执行时间：" + (System.currentTimeMillis( )- begin));
}
private List<String> getDuplicateElements(List<String> list) {
return list.stream()
.collect(Collectors.toMap(e -> e, e -> 1, Integer::sum))
.entrySet().stream()
.filter(entry -> entry.getValue() > 1)
.map(entry -> entry.getKey())
.collect(Collectors.toList());
}
}
