Skip to content

D
dc-java
Commit 224c00fd by wutong
Options

OMS,SKUMS, 调拨数据分页查询修改, 避免漏数据.

parent 2c4348c4
Showing with 106 additions and 92 deletions
data-base/base-sync-oms-order/src/main/java/com/bailuntec/job/OrderSyncJob.java
...@@ -35,81 +35,62 @@ public class OrderSyncJob extends PointJob { ...@@ -35,81 +35,62 @@ public class OrderSyncJob extends PointJob {
@Override @Override
public void executeJob(ShardingContext shardingContext, JobPointLog jobPointLog) { public void executeJob(ShardingContext shardingContext, JobPointLog jobPointLog) {
//如果页号为0, 就表示上次查完了, 这次要从总页数开始查, 如果页号 > 0, 就降序查, 防止漏单
if (jobPointLog.getPageIndex() == 0) {
jobPointLog.setPageIndex(requestBailunOrder(jobPointLog));
}
if (jobPointLog.getPageIndex() > 0) {
do {
requestBailunOrder(jobPointLog);
jobPointLog.setPageIndex(jobPointLog.getPageIndex() - 1);
if (jobPointLog.getPageIndex() % 10 == 0) {
try {
JobPointLogMapper mapper = SessionUtil.getSession().getMapper(JobPointLogMapper.class);
mapper.upsertSelective(jobPointLog);
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException("Mybatis操作DB失败", e);
} finally {
SessionUtil.closeSession();
}
}
} while (0 < jobPointLog.getPageIndex());
}
if (jobPointLog.getPageIndex() == 0) {
jobPointLog.setStartTime(jobPointLog.getEndTime());
jobPointLog.setEndTime(jobPointLog.getEndTime().plusDays(jobPointLog.getIntervalTime()).isAfter(LocalDateTime.now()) ? LocalDateTime.now() : jobPointLog.getEndTime().plusDays(jobPointLog.getIntervalTime()));
}
}
/**
* 请求OMS接口
*
* @param jobPointLog
* @return
*/
private Integer requestBailunOrder(JobPointLog jobPointLog) {
Integer totalPages = null;
LinkedHashMap<String, String> map = new LinkedHashMap<>(4); LinkedHashMap<String, String> map = new LinkedHashMap<>(4);
map.put("pageIndex", jobPointLog.getPageIndex() == 0 ? "1" : jobPointLog.getPageIndex().toString());
map.put("pageCount", jobPointLog.getPageSize().toString()); map.put("pageCount", jobPointLog.getPageSize().toString());
//时间回退一点, 避免服务器时间不一致而漏单 //时间回退一点, 避免服务器时间不一致而漏单
map.put("BailunLastUpdateTimeFrom", DateTimeFormatter.ofPattern(CommonConstant.TIME_FORMAT).format(jobPointLog.getStartTime().minusMinutes(5))); map.put("BailunLastUpdateTimeFrom", DateTimeFormatter.ofPattern(CommonConstant.TIME_FORMAT).format(jobPointLog.getStartTime().minusMinutes(3)));
map.put("BailunLastUpdateTimeTo", DateTimeFormatter.ofPattern(CommonConstant.TIME_FORMAT).format(jobPointLog.getEndTime())); map.put("BailunLastUpdateTimeTo", DateTimeFormatter.ofPattern(CommonConstant.TIME_FORMAT).format(jobPointLog.getEndTime()));
Response response = null; do {
String omsResultStr = null; map.put("pageIndex", jobPointLog.getPageIndex().equals(0) ? "1" : jobPointLog.getPageIndex().toString());
try { Response response = null;
Request request = new Request.Builder() String omsResultStr = null;
.get() try {
.url(OkHttpUtil.attachHttpGetParams(propertiesUtil.getPropertyAsString("BAILUNORDER_URL"), map)) Request request = new Request.Builder()
.addHeader("Content-Type", "application/json") .get()
.build(); .url(OkHttpUtil.attachHttpGetParams(propertiesUtil.getPropertyAsString("BAILUNORDER_URL"), map))
response = okHttpClient.newCall(request).execute(); .addHeader("Content-Type", "application/json")
omsResultStr = response.body().string(); .build();
} catch (IOException e) { response = okHttpClient.newCall(request).execute();
throw new RuntimeException(map + "请求OMS接口同步百伦接口失败" + response, e); omsResultStr = response.body().string();
} finally { } catch (IOException e) {
if (response != null) { throw new RuntimeException(map + "请求OMS接口同步百伦接口失败" + response, e);
response.close(); } finally {
if (response != null) {
response.close();
}
} }
} if (StringUtils.isNoneBlank(omsResultStr)) {
if (StringUtils.isNoneBlank(omsResultStr)) { OmsResultRoot omsResultRoot = JSON.parseObject(omsResultStr, OmsResultRoot.class);
OmsResultRoot omsResultRoot = JSON.parseObject(omsResultStr, OmsResultRoot.class); if (omsResultRoot != null && omsResultRoot.getSuccess().booleanValue()) {
if (omsResultRoot != null && omsResultRoot.getSuccess().booleanValue()) { OmsResultInfo omsResultInfo = omsResultRoot.getResult();
OmsResultInfo omsResultInfo = omsResultRoot.getResult(); if (jobPointLog.getPageIndex().equals(0)) {
totalPages = omsResultInfo.getTotalPages(); jobPointLog.setPageIndex(omsResultInfo.getTotalPages() + 1);
if (omsResultInfo.getResult() != null && omsResultInfo.getResult().size() > 0) { }
analyseOmsOrder(omsResultInfo.getResult()); if (omsResultInfo.getResult() != null && omsResultInfo.getResult().size() > 0) {
analyseOmsOrder(omsResultInfo.getResult());
}
} else {
throw new RuntimeException("调用OMS接口同步百伦订单失败, 响应200, 请求参数" + map.toString());
} }
} else { } else {
throw new RuntimeException("调用OMS接口同步百伦订单失败, 响应200, 请求参数" + map.toString()); throw new RuntimeException("调用OMS接口同步百伦订单失败, 响应为null, 请求参数" + map.toString());
} }
} else { if (jobPointLog.getPageIndex() % 10 == 0) {
throw new RuntimeException("调用OMS接口同步百伦订单失败, 响应为null, 请求参数" + map.toString()); try {
} JobPointLogMapper mapper = SessionUtil.getSession().getMapper(JobPointLogMapper.class);
//如果PageIndex等于0取总页数, 否则取PageIndex mapper.upsertSelective(jobPointLog);
return jobPointLog.getPageIndex() == 0 ? totalPages : jobPointLog.getPageIndex(); } catch (Exception e) {
e.printStackTrace();
throw new RuntimeException("Mybatis操作DB插入任务记录失败", e);
} finally {
SessionUtil.closeSession();
}
}
jobPointLog.setPageIndex(jobPointLog.getPageIndex() - 1);
} while (0 < jobPointLog.getPageIndex());
jobPointLog.setPageIndex(0);
jobPointLog.setStartTime(jobPointLog.getEndTime());
jobPointLog.setEndTime(jobPointLog.getEndTime().plusDays(jobPointLog.getIntervalTime()).isAfter(LocalDateTime.now()) ? LocalDateTime.now() : jobPointLog.getEndTime().plusDays(jobPointLog.getIntervalTime()));
} }
/** /**
......
data-base/base-sync-oms-order/src/main/resources/job.properties
...@@ -9,7 +9,7 @@ EVENT_RDB_STORAGE_USERNAME=root ...@@ -9,7 +9,7 @@ EVENT_RDB_STORAGE_USERNAME=root
EVENT_RDB_STORAGE_PASSWORD=#7kfnymAM$Y9-Ntf EVENT_RDB_STORAGE_PASSWORD=#7kfnymAM$Y9-Ntf
ZOOKEEPER_SERVER=172.31.255.120:2181 ZOOKEEPER_SERVER=172.31.255.120:2181
NAME_SPACE=data-center NAME_SPACE=data-center
JOB_NAME=base-sync-oms-order-coroutine #JOB_NAME=base-sync-oms-order-coroutine
#JOB_NAME=base-sync-oms-order JOB_NAME=base-sync-oms-order
JOB_CRON=0/1 * * * * ? * JOB_CRON=0/1 * * * * ? *
SHARDING_TOTAL_COUNT=1 SHARDING_TOTAL_COUNT=1
\ No newline at end of file
data-base/base-sync-sku/src/main/java/com/bailuntec/job/SkuMSSyncJob.java
...@@ -68,7 +68,7 @@ public class SkuMSSyncJob extends PointJob { ...@@ -68,7 +68,7 @@ public class SkuMSSyncJob extends PointJob {
handleSkuMsJson(data,jobPointLog); handleSkuMsJson(data,jobPointLog);
} }
if (jobPointLog.getPageIndex().equals(0) && result.getTotalPage() != null) { if (jobPointLog.getPageIndex().equals(0) && result.getTotalPage() != null) {
jobPointLog.setPageIndex(result.getTotalPage()); jobPointLog.setPageIndex(result.getTotalPage() + 1);
} }
} }
} else { } else {
......
data-base/base-sync-transfer/src/main/java/com/bailuntec/job/TransferDetailsServiceImpl.java
...@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON; ...@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import com.bailuntec.domain.constant.CommonConstant; import com.bailuntec.domain.constant.CommonConstant;
import com.bailuntec.domain.entity.*; import com.bailuntec.domain.entity.*;
import com.bailuntec.domain.example.*;
import com.bailuntec.domain.pojo.TransferDetailsPOJO; import com.bailuntec.domain.pojo.TransferDetailsPOJO;
import com.bailuntec.domain.pojo.TransferStreamPOJO; import com.bailuntec.domain.pojo.TransferStreamPOJO;
import com.bailuntec.mapper.*; import com.bailuntec.mapper.*;
...@@ -31,10 +32,8 @@ public class TransferDetailsServiceImpl { ...@@ -31,10 +32,8 @@ public class TransferDetailsServiceImpl {
map.put("EndDate", DateTimeFormatter.ofPattern(CommonConstant.TIME_FORMAT).format(jobPointLog.getEndTime())); map.put("EndDate", DateTimeFormatter.ofPattern(CommonConstant.TIME_FORMAT).format(jobPointLog.getEndTime()));
map.put("State", jobPointLog.getType()); map.put("State", jobPointLog.getType());
map.put("PageRow", jobPointLog.getPageSize()); map.put("PageRow", jobPointLog.getPageSize());
Integer page = 1;
Integer pagetotal = 0;
do { do {
map.put("CurrentPage", page); map.put("CurrentPage", jobPointLog.getPageIndex() > 0?jobPointLog.getPageIndex() : 1);
Response response = null; Response response = null;
String responseStr = null; String responseStr = null;
try { try {
...@@ -57,6 +56,9 @@ public class TransferDetailsServiceImpl { ...@@ -57,6 +56,9 @@ public class TransferDetailsServiceImpl {
JSONObject jsonObject = JSON.parseObject(responseStr); JSONObject jsonObject = JSON.parseObject(responseStr);
if (jsonObject.get("isSuccess") != null && jsonObject.getBooleanValue("isSuccess")) { if (jsonObject.get("isSuccess") != null && jsonObject.getBooleanValue("isSuccess")) {
TransferDetailsPOJO transferDetailsPOJO = jsonObject.getObject("data", TransferDetailsPOJO.class); TransferDetailsPOJO transferDetailsPOJO = jsonObject.getObject("data", TransferDetailsPOJO.class);
if (jobPointLog.getPageIndex().equals(0)) {
jobPointLog.setPageIndex(transferDetailsPOJO.getPageCount() + 1);
}
if (transferDetailsPOJO != null && transferDetailsPOJO.getData() != null && transferDetailsPOJO.getData().size() > 0) { if (transferDetailsPOJO != null && transferDetailsPOJO.getData() != null && transferDetailsPOJO.getData().size() > 0) {
switch (jobPointLog.getType()) { switch (jobPointLog.getType()) {
case 1: case 1:
...@@ -69,8 +71,11 @@ public class TransferDetailsServiceImpl { ...@@ -69,8 +71,11 @@ public class TransferDetailsServiceImpl {
} }
try { try {
DcBaseTransferBaleMapper mapper1 = SessionUtil.getSession().getMapper(DcBaseTransferBaleMapper.class); DcBaseTransferBaleMapper mapper1 = SessionUtil.getSession().getMapper(DcBaseTransferBaleMapper.class);
mapper1.upsertSelective(dcBaseTransferBale); int i = mapper1.updateByExampleSelective(dcBaseTransferBale, DcBaseTransferBaleExample.newAndCreateCriteria().andBailunSkuEqualTo(dcBaseTransferBale.getBailunSku())
SessionUtil.getSession().commit(); .andTransferOrderIdEqualTo(dcBaseTransferBale.getTransferOrderId()).example());
if (i == 0) {
mapper1.insertSelective(dcBaseTransferBale);
}
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
throw new RuntimeException("MYbatis操作DB失败", e); throw new RuntimeException("MYbatis操作DB失败", e);
...@@ -89,8 +94,11 @@ public class TransferDetailsServiceImpl { ...@@ -89,8 +94,11 @@ public class TransferDetailsServiceImpl {
} }
try { try {
DcBaseTransferDeliveryMapper mapper2 = SessionUtil.getSession().getMapper(DcBaseTransferDeliveryMapper.class); DcBaseTransferDeliveryMapper mapper2 = SessionUtil.getSession().getMapper(DcBaseTransferDeliveryMapper.class);
mapper2.upsertSelective(dcBaseTransferDelivery); int i = mapper2.updateByExampleSelective(dcBaseTransferDelivery, DcBaseTransferDeliveryExample.newAndCreateCriteria().andBailunSkuEqualTo(dcBaseTransferDelivery.getBailunSku())
SessionUtil.getSession().commit(); .andTransferOrderIdEqualTo(dcBaseTransferDelivery.getTransferOrderId()).example());
if (i == 0) {
mapper2.insertSelective(dcBaseTransferDelivery);
}
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
throw new RuntimeException("MYbatis操作DB失败", e); throw new RuntimeException("MYbatis操作DB失败", e);
...@@ -110,8 +118,11 @@ public class TransferDetailsServiceImpl { ...@@ -110,8 +118,11 @@ public class TransferDetailsServiceImpl {
} }
try { try {
DcBaseTransferVerifyMapper mapper3 = SessionUtil.getSession().getMapper(DcBaseTransferVerifyMapper.class); DcBaseTransferVerifyMapper mapper3 = SessionUtil.getSession().getMapper(DcBaseTransferVerifyMapper.class);
mapper3.upsertSelective(dcBaseTransferVerify); int i = mapper3.updateByExampleSelective(dcBaseTransferVerify, DcBaseTransferVerifyExample.newAndCreateCriteria().andBailunSkuEqualTo(dcBaseTransferVerify.getBailunSku())
SessionUtil.getSession().commit(); .andTransferOrderIdEqualTo(dcBaseTransferVerify.getTransferOrderId()).example());
if (i == 0) {
mapper3.insertSelective(dcBaseTransferVerify);
}
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
throw new RuntimeException("MYbatis操作DB失败", e); throw new RuntimeException("MYbatis操作DB失败", e);
...@@ -130,8 +141,11 @@ public class TransferDetailsServiceImpl { ...@@ -130,8 +141,11 @@ public class TransferDetailsServiceImpl {
} }
try { try {
DcBaseTransferTransitMapper mapper4 = SessionUtil.getSession().getMapper(DcBaseTransferTransitMapper.class); DcBaseTransferTransitMapper mapper4 = SessionUtil.getSession().getMapper(DcBaseTransferTransitMapper.class);
mapper4.upsertSelective(dcBaseTransferTransit); int i = mapper4.updateByExampleSelective(dcBaseTransferTransit, DcBaseTransferTransitExample.newAndCreateCriteria().andBailunSkuEqualTo(dcBaseTransferTransit.getBailunSku())
SessionUtil.getSession().commit(); .andTransferOrderIdEqualTo(dcBaseTransferTransit.getTransferOrderId()).example());
if (i == 0) {
mapper4.insertSelective(dcBaseTransferTransit);
}
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
throw new RuntimeException("MYbatis操作DB失败", e); throw new RuntimeException("MYbatis操作DB失败", e);
...@@ -150,8 +164,11 @@ public class TransferDetailsServiceImpl { ...@@ -150,8 +164,11 @@ public class TransferDetailsServiceImpl {
} }
try { try {
DcBaseTransferCompleteMapper mapper5 = SessionUtil.getSession().getMapper(DcBaseTransferCompleteMapper.class); DcBaseTransferCompleteMapper mapper5 = SessionUtil.getSession().getMapper(DcBaseTransferCompleteMapper.class);
mapper5.upsertSelective(dcBaseTransferComplete); int i = mapper5.updateByExampleSelective(dcBaseTransferComplete, DcBaseTransferCompleteExample.newAndCreateCriteria().andBailunSkuEqualTo(dcBaseTransferComplete.getBailunSku())
SessionUtil.getSession().commit(); .andTransferOrderIdEqualTo(dcBaseTransferComplete.getTransferOrderId()).example());
if (i == 0) {
mapper5.insertSelective(dcBaseTransferComplete);
}
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
throw new RuntimeException("MYbatis操作DB失败", e); throw new RuntimeException("MYbatis操作DB失败", e);
...@@ -170,8 +187,11 @@ public class TransferDetailsServiceImpl { ...@@ -170,8 +187,11 @@ public class TransferDetailsServiceImpl {
} }
try { try {
DcBaseTransferInboundMapper mapper12 = SessionUtil.getSession().getMapper(DcBaseTransferInboundMapper.class); DcBaseTransferInboundMapper mapper12 = SessionUtil.getSession().getMapper(DcBaseTransferInboundMapper.class);
mapper12.upsertSelective(dcBaseTransferInbound); int i = mapper12.updateByExampleSelective(dcBaseTransferInbound, DcBaseTransferInboundExample.newAndCreateCriteria().andBailunSkuEqualTo(dcBaseTransferInbound.getBailunSku())
SessionUtil.getSession().commit(); .andTransferOrderIdEqualTo(dcBaseTransferInbound.getTransferOrderId()).example());
if (i == 0) {
mapper12.insertSelective(dcBaseTransferInbound);
}
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
throw new RuntimeException("MYbatis操作DB失败", e); throw new RuntimeException("MYbatis操作DB失败", e);
...@@ -182,14 +202,26 @@ public class TransferDetailsServiceImpl { ...@@ -182,14 +202,26 @@ public class TransferDetailsServiceImpl {
break; break;
} }
} }
jobPointLog.setStartTime(jobPointLog.getEndTime()); } else {
jobPointLog.setEndTime(jobPointLog.getStartTime().plusDays(jobPointLog.getIntervalTime().longValue()).isAfter(LocalDateTime.now()) ? LocalDateTime.now() : jobPointLog.getStartTime().plusDays(jobPointLog.getIntervalTime().longValue())); throw new RuntimeException("调用调拨流水接口返回200但是失败");
} }
} else { } else {
throw new RuntimeException("调用调拨流水接口失败"); throw new RuntimeException("调用调拨流水接口失败");
} }
page++; if (jobPointLog.getPageIndex() % 5 == 0) {
} while (page <= pagetotal); try {
JobPointLogMapper mapper = SessionUtil.getSession().getMapper(JobPointLogMapper.class);
mapper.upsertSelective(jobPointLog);
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException("Mybatis操作DB插入任务记录失败", e);
} finally {
SessionUtil.closeSession();
}
}
jobPointLog.setPageIndex(jobPointLog.getPageIndex() - 1);
} while (jobPointLog.getPageIndex() > 0);
jobPointLog.setPageIndex(0);
jobPointLog.setStartTime(jobPointLog.getEndTime()); jobPointLog.setStartTime(jobPointLog.getEndTime());
jobPointLog.setEndTime(jobPointLog.getStartTime().plusDays(jobPointLog.getIntervalTime().longValue()).isAfter(LocalDateTime.now()) ? LocalDateTime.now() : jobPointLog.getStartTime().plusDays(jobPointLog.getIntervalTime().longValue())); jobPointLog.setEndTime(jobPointLog.getStartTime().plusDays(jobPointLog.getIntervalTime().longValue()).isAfter(LocalDateTime.now()) ? LocalDateTime.now() : jobPointLog.getStartTime().plusDays(jobPointLog.getIntervalTime().longValue()));
} }
......
data-base/base-sync-transfer/src/test/java/TransferDetailsTest.java
...@@ -8,9 +8,9 @@ public class TransferDetailsTest { ...@@ -8,9 +8,9 @@ public class TransferDetailsTest {
@Test @Test
public void test() { public void test() {
JobPointLog jobPointLog = new JobPointLog(); JobPointLog jobPointLog = new JobPointLog();
jobPointLog.setStartTime(LocalDateTime.of(2019,02,12,17,13,18)); jobPointLog.setStartTime(LocalDateTime.of(2019,04,17,14,53,14));
jobPointLog.setEndTime(LocalDateTime.of(2019,02,13,17,13,19)); jobPointLog.setEndTime(LocalDateTime.of(2019,04,17,14,53,15));
jobPointLog.setType(3); jobPointLog.setType(5);
jobPointLog.setIntervalTime(1); jobPointLog.setIntervalTime(1);
jobPointLog.setPageIndex(1); jobPointLog.setPageIndex(1);
jobPointLog.setPageSize(1000); jobPointLog.setPageSize(1000);
......
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