DealGeoInOutParkDataSchedule.java
5.17 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
package com.zteits.park.configuration.schedule;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import com.clouds.common.utils.ExceptionUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSON;
import com.clouds.common.cache.sys.SysCodeValueCacheUtil;
import com.clouds.common.constants.CodeValKindEnum;
import com.clouds.common.constants.ParkConstant;
import com.clouds.common.utils.IPUtil;
import com.zteits.clouds.api.apibase.constants.ErrorType;
import com.zteits.clouds.api.apibase.constants.ParkBusiType;
import com.zteits.clouds.api.apibase.exception.BizException;
import com.zteits.clouds.api.dto.rocketmq.datacollection.georeport.GeoInOutReportMsgVO;
import com.zteits.park.configuration.schedule.param.DistributedTaskPropVO;
import com.zteits.park.domain.TpPInterfaceLogDO;
import com.zteits.park.domain.TpPInterfaceLogWithBLOBs;
import com.zteits.park.mapper.TpPInterfaceLogMapper;
import com.zteits.park.mapper.smapper.TpPInterfaceLogSMapper;
import com.zteits.park.service.eqpinoutrecord.oper.GeoInOutParkReportInnerService;
/**
* 处理地磁进出场上报数据
* .<br/>
*
* Copyright: Copyright (c) 2017 zteits
*
* @ClassName: DealGeoInOutParkDataSchedule
* @Description:
* @version: v1.0.0
* @author: zhaowg
* @date: 2018年5月4日 下午3:11:36
* Modification History:
* Date Author Version Description
*---------------------------------------------------------*
* 2018年5月4日 zhaowg v1.0.0 创建
*/
@Component
public class DealGeoInOutParkDataSchedule{
private static final Logger logger = LoggerFactory.getLogger(DealGeoInOutParkDataSchedule.class);
@Autowired
private GeoInOutParkReportInnerService geoInOutParkReportInnerService;
@Autowired
private TpPInterfaceLogSMapper interfaceLogSMapper;
@Autowired
private TpPInterfaceLogMapper interfaceLogMapper;
/**
* 间隔3秒执行一次
*
* 2018年5月4日 zhaowg
*/
@Scheduled(fixedDelay=1000)
public void dealGeoInOutData(){
//获取分布式任务配置信息
String distributedTaskProp = SysCodeValueCacheUtil.queryCodeValue(CodeValKindEnum.DISTRIBUTED_TASK_PROP);
if(StringUtils.isBlank(distributedTaskProp)){
logger.debug("未配置分布式任务相关信息，不处理");
return;
}
//获取当前服务器IP
String ip = IPUtil.getLocalHost();
List<DistributedTaskPropVO> taskVOs = JSON.parseArray(distributedTaskProp, DistributedTaskPropVO.class);
Map<String, Integer> map = taskVOs.stream().collect(Collectors.toMap(DistributedTaskPropVO::getIp, DistributedTaskPropVO::getDelivery));
if(!map.containsKey(ip)){
logger.debug("分布式任务配置得服务器中不包含该服务器："+ip+",故该服务器不处理任何数据");
return;
}
Integer modelNum = map.get(ip);
if(modelNum==null){
logger.debug("分布式任务，该服务器："+ip+"配置的取模参数为null,不处理任务数据");
return;
}
int allNum = map.size();
logger.debug("分布式任务，服务器一共："+allNum+"台，当前服务器ip:"+ip+"，处理的取模数据为："+modelNum);
this.execute(allNum,modelNum);
}
private void execute(int allNum, Integer modelNum) {
//查询需要该服务器处理的数据
TpPInterfaceLogDO interfaceLogDO = new TpPInterfaceLogDO();
interfaceLogDO.setAllNum(allNum);
interfaceLogDO.setBusiType(ParkBusiType.UPDATE_BERTH_STATUS_BY_EQP.getValue());
interfaceLogDO.setDealStatus(ParkConstant.DealState.WAIT_DEAL);
interfaceLogDO.setModelNum(modelNum);
List<TpPInterfaceLogWithBLOBs> logWithBLOBs = interfaceLogSMapper.selectByModel(interfaceLogDO );
if(CollectionUtils.isEmpty(logWithBLOBs)){
logger.debug("未查询到地磁上报数据");
return;
}
for (TpPInterfaceLogWithBLOBs tpPInterfaceLogWithBLOBs : logWithBLOBs) {
try{
//判断该条记录和当前时间相差是否小于30分钟，超过30分钟不处理
if(new Date().getTime()-tpPInterfaceLogWithBLOBs.getCreateTime().getTime()>30*60*1000){
throw new BizException(ErrorType.BIZ_ERROR,"超过30分钟的消息，不处理",false);
}
GeoInOutReportMsgVO msgVO = JSON.parseObject(tpPInterfaceLogWithBLOBs.getRequestBody(), GeoInOutReportMsgVO.class);
geoInOutParkReportInnerService.dealGeoInOutParkReportData(msgVO);
//删除该记录
interfaceLogMapper.deleteByPrimaryKey(tpPInterfaceLogWithBLOBs.getId());
}catch(Exception e){
logger.error("处理地磁失败，原始信息为："+tpPInterfaceLogWithBLOBs.getRequestBody(),e);
TpPInterfaceLogWithBLOBs upInterface = new TpPInterfaceLogWithBLOBs();
upInterface.setId(tpPInterfaceLogWithBLOBs.getId());
upInterface.setDealStatus(ParkConstant.DealState.FAIL);
upInterface.setErrorMsg(ExceptionUtil.getTrace(e));
upInterface.setModifyTime(new Date());
interfaceLogMapper.updateByPrimaryKeySelective(upInterface);
}
}
}
}