DealGeoInOutParkDataSchedule.java
5.53 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
126
127
128
129
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.boot.autoconfigure.condition.ConditionalOnProperty;
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
// 存量定时任务总开关：新老服务同时在线时默认关闭，避免同一份数据被处理两遍。
// 切换步骤见 MIGRATION.md「§7.5 存量定时任务去重」。
@ConditionalOnProperty(prefix = "parking.legacy.scheduler", name = "enabled", havingValue = "true")
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);
}
}
}
}