Skip to content

B
bigdatainterservice
Commit cdf8b523 by wangling
Options

Merge remote-tracking branch 'origin/master'

parents 438dee73 2a67f58c
Showing with 194 additions and 10 deletions
src/main/java/com/founder/interservice/mapper/xzxt/TrackTogetherMapper.java
...@@ -20,4 +20,6 @@ public interface TrackTogetherMapper { ...@@ -20,4 +20,6 @@ public interface TrackTogetherMapper {
* @return * @return
*/ */
public List<TrackTogetherTask> queryTasksByAsjbhAndFwbsh(TrackTogetherTask taskParam); public List<TrackTogetherTask> queryTasksByAsjbhAndFwbsh(TrackTogetherTask taskParam);
int queryTasksByAsjbhAndFwbshCount(TrackTogetherTask taskParam);
} }
src/main/java/com/founder/interservice/regionalanalysis/service/impl/RegionalsScheduledService.java
...@@ -238,7 +238,7 @@ public class RegionalsScheduledService { ...@@ -238,7 +238,7 @@ public class RegionalsScheduledService {
/** /**
* 通过区域碰撞的结果 使用第四个接口调取关联数据(原生方法) * 通过区域碰撞的结果 使用第四个接口调取关联数据(原生方法)
*/ */
public void getAndSaveRydetail(List<RegionalsTaskResult> resultsList) { public void getAndSaveRydetail(List<RegionalsTaskResult> resultsList) throws Exception{
if (resultsList != null && !resultsList.isEmpty()) { if (resultsList != null && !resultsList.isEmpty()) {
for (RegionalsTaskResult result : resultsList) { for (RegionalsTaskResult result : resultsList) {
String objType = result.getObjectType(); String objType = result.getObjectType();
......
src/main/java/com/founder/interservice/tracktraveltogether/controller/TrackTogetherContoller.java
...@@ -169,6 +169,7 @@ public class TrackTogetherContoller { ...@@ -169,6 +169,7 @@ public class TrackTogetherContoller {
taskParam.setBegin(begin); taskParam.setBegin(begin);
taskParam.setEnd(end); taskParam.setEnd(end);
List<TrackTogetherTask> togetherTaskLsit = trackTogetherService.queryTasksByAsjbhAndFwbsh(taskParam); List<TrackTogetherTask> togetherTaskLsit = trackTogetherService.queryTasksByAsjbhAndFwbsh(taskParam);
int count = trackTogetherService.queryTasksByAsjbhAndFwbshCount(taskParam);
if(null != togetherTaskLsit && !togetherTaskLsit.isEmpty()){ if(null != togetherTaskLsit && !togetherTaskLsit.isEmpty()){
togetherTaskVOS = new ArrayList<>(); togetherTaskVOS = new ArrayList<>();
for (TrackTogetherTask task : togetherTaskLsit) { for (TrackTogetherTask task : togetherTaskLsit) {
...@@ -207,7 +208,10 @@ public class TrackTogetherContoller { ...@@ -207,7 +208,10 @@ public class TrackTogetherContoller {
togetherTaskVO.setState("运行中"); togetherTaskVO.setState("运行中");
break; break;
case "FINISHED": case "FINISHED":
togetherTaskVO.setState("已完成"); togetherTaskVO.setState("标识号调取完成");
break;
case "END":
togetherTaskVO.setState("伴随调取完成");
break; break;
case "TIMEOUT": case "TIMEOUT":
togetherTaskVO.setState("计算超时"); togetherTaskVO.setState("计算超时");
...@@ -224,7 +228,7 @@ public class TrackTogetherContoller { ...@@ -224,7 +228,7 @@ public class TrackTogetherContoller {
}else{ }else{
togetherTaskVOS = new ArrayList<>(); togetherTaskVOS = new ArrayList<>();
} }
resultMap.put("total", togetherTaskVOS.size()); resultMap.put("total", count);
resultMap.put("rows",togetherTaskVOS); resultMap.put("rows",togetherTaskVOS);
}catch (InterServiceException e){ }catch (InterServiceException e){
e.printStackTrace(); e.printStackTrace();
......
src/main/java/com/founder/interservice/tracktraveltogether/service/TrackTogetherService.java
...@@ -27,6 +27,8 @@ public interface TrackTogetherService { ...@@ -27,6 +27,8 @@ public interface TrackTogetherService {
public List<TrackTogetherTask> queryTasksByAsjbhAndFwbsh(TrackTogetherTask taskParam); public List<TrackTogetherTask> queryTasksByAsjbhAndFwbsh(TrackTogetherTask taskParam);
public int queryTasksByAsjbhAndFwbshCount(TrackTogetherTask taskParam);
public List<TogetherTaskResult> getTogetherTaskResultList(TogetherTaskResult taskParam); public List<TogetherTaskResult> getTogetherTaskResultList(TogetherTaskResult taskParam);
int getTogetherTaskResultListTotalCount(TogetherTaskResult taskParam); int getTogetherTaskResultListTotalCount(TogetherTaskResult taskParam);
......
src/main/java/com/founder/interservice/tracktraveltogether/service/impl/TogetherScheduledService.java
...@@ -4,6 +4,8 @@ import com.alibaba.fastjson.JSONArray; ...@@ -4,6 +4,8 @@ import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import com.founder.interservice.model.AutoTbStRy; import com.founder.interservice.model.AutoTbStRy;
import com.founder.interservice.regionalanalysis.model.RegionalTaskResultDetail; import com.founder.interservice.regionalanalysis.model.RegionalTaskResultDetail;
import com.founder.interservice.regionalanalysis.model.RegionalsTaskResult;
import com.founder.interservice.regionalanalysis.service.impl.RegionalsScheduledService;
import com.founder.interservice.service.IphoneTrackService; import com.founder.interservice.service.IphoneTrackService;
import com.founder.interservice.service.PushWaDataService; import com.founder.interservice.service.PushWaDataService;
import com.founder.interservice.tracktraveltogether.model.TogetherTaskResult; import com.founder.interservice.tracktraveltogether.model.TogetherTaskResult;
...@@ -61,13 +63,17 @@ public class TogetherScheduledService { ...@@ -61,13 +63,17 @@ public class TogetherScheduledService {
* @Author: cao peng * @Author: cao peng
* @date: 2018/8/22 0022-16:35 * @date: 2018/8/22 0022-16:35
*/ */
@Scheduled(initialDelay = 120000,fixedDelay = 60000) //项目启动后延迟三分钟秒执行，每次执行完后三分钟后再次执行
/**
* @Description: 二次优化，调整如下,
* 1.第一步，首先调取伴随轨迹数据
* 2.第二步，调取伴随结果，使用批处理调取关联关系
* @Author: liuys
* 此为第一步，调取伴随任务的电子数据
*/
@Scheduled(initialDelay = 100000,fixedDelay = 60000) //项目启动后延迟三分钟秒执行，每次执行完后三分钟后再次执行
public void queryTaskResult(){ public void queryTaskResult(){
System.out.println("=============TogetherTask Start Running================"); System.out.println("=============TogetherTask Start Running================");
System.out.println("=============TogetherTask Start Running================");
System.out.println("=============TogetherTask Start Running================");
System.out.println("=============TogetherTask Start Running================");
System.out.println("=============TogetherTask Start Running================");
try{ try{
//1 下去查询任务表中status = "QUEUEING","STARTING","RUNNING"的任务 //1 下去查询任务表中status = "QUEUEING","STARTING","RUNNING"的任务
List<TrackTogetherTask> taskList = taskRepository.findAll(new Specification<TrackTogetherTask>() { List<TrackTogetherTask> taskList = taskRepository.findAll(new Specification<TrackTogetherTask>() {
...@@ -133,10 +139,10 @@ public class TogetherScheduledService { ...@@ -133,10 +139,10 @@ public class TogetherScheduledService {
List<TogetherTaskResult> results = taskResultRepository.findAll(example); List<TogetherTaskResult> results = taskResultRepository.findAll(example);
if (results == null || results.isEmpty()) { if (results == null || results.isEmpty()) {
taskResultRepository.save(r); taskResultRepository.save(r);
getAndSaveRydetail(taskResults); //getAndSaveRydetail(taskResults);
} }
} }
getAndSaveRydetail(taskResults); //getAndSaveRydetail(taskResults);
} }
} }
} }
...@@ -147,6 +153,46 @@ public class TogetherScheduledService { ...@@ -147,6 +153,46 @@ public class TogetherScheduledService {
} }
} }
/**
* @Description: 第二步，开启任务，扫描任务表，调取关联关系，然后更新
* @Author: liuys
*/
@Scheduled(initialDelay = 120000,fixedDelay = 60000)
public void queryFinishTaskResult(){
try{
List<TrackTogetherTask> taskList = taskRepository.findAll(new Specification<TrackTogetherTask>() {
@Override
public Predicate toPredicate(Root<TrackTogetherTask> root, CriteriaQuery<?> criteriaQuery, CriteriaBuilder criteriaBuilder) {
Predicate p1 = criteriaBuilder.like(root.get("taskCaseId").as(String.class), "A%");
Predicate p2 = criteriaBuilder.equal(root.get("state").as(String.class), "FINISHED");
criteriaQuery.where(criteriaBuilder.and(p1,p2));
return criteriaQuery.getRestriction();
}
});
if(taskList != null && !taskList.isEmpty()){
for (TrackTogetherTask task:taskList){
List<TogetherTaskResult> taskResults = taskResultRepository.findAllByTaskId(task.getTaskId());
//使用批处理调取关联关系
if(taskResults!=null&&!taskResults.isEmpty()){
int limit = 50;//批处理量
int size = taskResults.size();
long threadNum = size / limit;
for (int i = 0; i < threadNum + 1; i++) {
TogetherScheduledService.HandleThreadGlsj thread = new TogetherScheduledService.HandleThreadGlsj(taskResults, i * limit, ((i + 1) * limit));
thread.start();
}
}
updateTaskStates(task.getTaskId(),"2","END"); //修改任务状态
}
}
}catch (Exception e){
e.printStackTrace();
}
}
/** /**
* 拿取到伴随的结果后 使用第四个接口调取关联数据 * 拿取到伴随的结果后 使用第四个接口调取关联数据
*/ */
...@@ -268,6 +314,22 @@ public class TogetherScheduledService { ...@@ -268,6 +314,22 @@ public class TogetherScheduledService {
return list; return list;
} }
/* public static void main(String[] args){
List<TogetherTaskResult> list = new ArrayList<>();
for (int i=0;i<1000;i++){
TogetherTaskResult taskResult = new TogetherTaskResult();
taskResult.setXXZJBH(i+"zz");
taskResult.setObjectValue(i/3+"cc");
taskResult.setObjectType(i/3+"dd");
taskResult.setCount(i/3);
list.add(taskResult);
}
System.out.println(list.size());
list = removeDuplicate(list);
System.out.println(list.size());
}*/
/** /**
* 调用新版接口获取任务状态 * 调用新版接口获取任务状态
* @param taskId * @param taskId
...@@ -290,4 +352,95 @@ public class TogetherScheduledService { ...@@ -290,4 +352,95 @@ public class TogetherScheduledService {
map.put("state",state); map.put("state",state);
return map; return map;
} }
/**
* 多线程批处理(getAndSaveRydetail)
* 使用批处理
* by liuys
*/
class HandleThreadGlsj extends Thread {
private List<TogetherTaskResult> data;
private int start;
private int end;
public HandleThreadGlsj(List<TogetherTaskResult> data, int start, int end) {
this.data = data;
this.start = start;
this.end = end;
}
public void run() {
if (start > data.size()) {
return;
}
if (end > data.size()) {
end = data.size();
}
//begin
List<TogetherTaskResult> resultsList = data.subList(start, end); //从fromIndex（包含），到toIndex（不包含）
try {
for (TogetherTaskResult result : resultsList) {
String objType = result.getObjectType();
String objValue = result.getObjectValue();
String objName = result.getObjectTypeName();
if (objValue != null && !"".equals(objValue)) {
//MAC地址、用户名、淘宝旺旺号不去调取关联数据，其它类型都使用第四个接口调取
if (!Arrays.asList("4329", "4342","45282").contains(objType)) {
String results =pushWaDataService.pushRelationAll(DateUtil.getYYYYMMDD(),DateUtil.getYYYYMMDDHHMM(),"admin",objValue,3);
JSONObject jsonObj = new JSONObject();
if(null != results && results.startsWith("{")){
jsonObj = JSONObject.parseObject(results);
}
if (jsonObj != null && !jsonObj.isEmpty()) {
JSONObject dataObj = jsonObj.getJSONObject("data");
RegionalTaskResultDetail detail = new RegionalTaskResultDetail();
detail.setTaskResultId(result.getXXZJBH());
detail.setXXZJBH(KeyUtil.getUniqueKey("EDZ"));
detail.setObjectType(objType);
detail.setObjectValue(objValue);
detail.setObjectTypeName(objName);
if(null != dataObj){
String objType2 = dataObj.getString("objType");
String objValue2 = dataObj.getString("objValue");
if("1".equals(objType2) && !objValue2.isEmpty()){
detail.setZjlxCode("004");
detail.setZjlx("身份证号");
detail.setZjhm(objValue2);
detail = getEdzxxForSfzh(detail); //获取二代证信息
//根据身份证判断是否存在 如果存在更新 不存在新增
RegionalTaskResultDetail param = new RegionalTaskResultDetail();
param.setObjectType(objType);
param.setObjectValue(objValue);
boolean isExist = togetherService.togetherDetailIsExist(param);
if (!isExist) {
togetherService.saveTogetherDetail(detail);
} else {
togetherService.updateTogetherDetail(detail);
}
}else if(Arrays.asList("20","4394","3996","6003").contains(objType2)){
//手机号码
detail.setSjhm(objValue2);
RegionalTaskResultDetail param = new RegionalTaskResultDetail();
param.setObjectType(objType);
param.setObjectValue(objValue);
boolean isExist = togetherService.togetherDetailIsExist(param);
if (!isExist) {
togetherService.saveTogetherDetail(detail);
} else {
togetherService.updateTogetherDetail(detail);
}
}
}
}
}
}
}
}catch(Exception e){
e.printStackTrace();
}
}
}
} }
src/main/java/com/founder/interservice/tracktraveltogether/service/impl/TrackTogetherServiceImpl.java
...@@ -164,6 +164,11 @@ public class TrackTogetherServiceImpl implements TrackTogetherService { ...@@ -164,6 +164,11 @@ public class TrackTogetherServiceImpl implements TrackTogetherService {
} }
@Override @Override
public int queryTasksByAsjbhAndFwbshCount(TrackTogetherTask taskParam) {
return trackTogetherMapper.queryTasksByAsjbhAndFwbshCount(taskParam);
}
@Override
public List<TogetherTaskResult> getTogetherTaskResultList(TogetherTaskResult taskParam) { public List<TogetherTaskResult> getTogetherTaskResultList(TogetherTaskResult taskParam) {
return togetherTaskResultMapper.getTogetherTaskResultList(taskParam); return togetherTaskResultMapper.getTogetherTaskResultList(taskParam);
} }
......
src/main/resources/mybatis/mapper/xzxt/TrackTogetkerTaskMapper.xml
...@@ -38,4 +38,21 @@ ...@@ -38,4 +38,21 @@
) t2 ) t2
where rn <![CDATA[ > ]]> #{begin } where rn <![CDATA[ > ]]> #{begin }
</select> </select>
<select id="queryTasksByAsjbhAndFwbshCount" parameterType="com.founder.interservice.tracktraveltogether.model.TrackTogetherTask" resultType="java.lang.Integer">
select
count(*)
from TB_ST_TRACKTOGETHERTASK t
<where>
<if test="taskCaseId != null and taskCaseId != ''">
t.TASK_CASE_ID = #{taskCaseId}
</if>
<if test="objectValue != null and objectValue != ''">
and t.OBJECT_VALUE = #{objectValue}
</if>
<if test="taskId != null and taskId != ''">
and t.TASK_ID = #{taskId}
</if>
</where>
</select>
</mapper> </mapper>
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment