订单数据源增加订单对比功能
This commit is contained in:
wuweihua
@@ -0,0 +1,211 @@
|
||||
package com.bms.price.task;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashSet;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Component;
|
||||
import com.bms.common.core.redis.RedisCache;
|
||||
import com.bms.common.utils.StringUtils;
|
||||
import com.bms.price.domain.vo.TabOrderCompareTask;
|
||||
import com.bms.price.domain.vo.TabOrderMonitorQuery;
|
||||
import com.bms.price.mapper.PrcOrderBillMapper;
|
||||
import com.bms.price.oracle.TabOrderMonitorService;
|
||||
import com.bms.price.search.IBillSearchStore;
|
||||
import com.bms.price.search.TabOrderSyncEsLog;
|
||||
|
||||
/**
|
||||
* 订单比对异步任务：分页拉上游转单号，与本地 transfer_no 比对差异。
|
||||
*/
|
||||
@Component
|
||||
public class TabOrderCompareAsyncRunner
|
||||
{
|
||||
private static final Logger log = LoggerFactory.getLogger(TabOrderCompareAsyncRunner.class);
|
||||
|
||||
public static final String CACHE_KEY_PREFIX = "billing:orderSource:compare:";
|
||||
public static final int TTL_MINUTES = 120;
|
||||
public static final int PAGE_SIZE = 500;
|
||||
/** 单次任务最多扫描上游转单号数（防崩） */
|
||||
public static final int SCAN_HARD_LIMIT = 50000;
|
||||
/** 进度/抽屉中保留的缺失转单号上限 */
|
||||
public static final int MISSING_SAMPLE_LIMIT = 5000;
|
||||
|
||||
@Autowired
|
||||
private TabOrderMonitorService tabOrderMonitorService;
|
||||
|
||||
@Autowired
|
||||
private PrcOrderBillMapper prcOrderBillMapper;
|
||||
|
||||
@Autowired
|
||||
private RedisCache redisCache;
|
||||
|
||||
@Autowired
|
||||
private IBillSearchStore billSearchStore;
|
||||
|
||||
@Async("threadPoolTaskExecutor")
|
||||
public void run(String asyncTaskId, TabOrderMonitorQuery query)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (query != null)
|
||||
{
|
||||
query.setOrderIdentifiers(null);
|
||||
query.setOrderId(null);
|
||||
query.setOrgOrderId(null);
|
||||
query.setTransferOrderId(null);
|
||||
}
|
||||
put(asyncTaskId, TabOrderCompareTask.STATUS_RUNNING, 0, "starting", 0, 0, 0, null, null);
|
||||
int scanned = 0;
|
||||
int matched = 0;
|
||||
int missing = 0;
|
||||
List<String> missingSample = new ArrayList<String>();
|
||||
int pageNum = 1;
|
||||
while (true)
|
||||
{
|
||||
if (scanned >= SCAN_HARD_LIMIT)
|
||||
{
|
||||
log.warn("orderSource compare capped at {} asyncTaskId={}", Integer.valueOf(SCAN_HARD_LIMIT), asyncTaskId);
|
||||
break;
|
||||
}
|
||||
int remain = SCAN_HARD_LIMIT - scanned;
|
||||
int pageSize = Math.min(PAGE_SIZE, remain);
|
||||
List<String> transferNos = tabOrderMonitorService.selectCompareTransferPage(query, pageNum, pageSize);
|
||||
if (transferNos == null || transferNos.isEmpty())
|
||||
{
|
||||
break;
|
||||
}
|
||||
Set<String> existing = findExisting(transferNos, query == null ? null : query.getCustomerCodes());
|
||||
for (String tn : transferNos)
|
||||
{
|
||||
if (StringUtils.isEmpty(tn))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
scanned++;
|
||||
String key = tn.trim().toUpperCase(Locale.ROOT);
|
||||
if (existing.contains(key))
|
||||
{
|
||||
matched++;
|
||||
}
|
||||
else
|
||||
{
|
||||
missing++;
|
||||
if (missingSample.size() < MISSING_SAMPLE_LIMIT)
|
||||
{
|
||||
missingSample.add(tn.trim());
|
||||
}
|
||||
}
|
||||
}
|
||||
int percent = Math.min(99, (int) ((scanned * 100L) / Math.max(scanned + pageSize, 1)));
|
||||
// 最后一页不满页时近乎完成；否则按已扫描量给一个缓慢增长的百分比
|
||||
if (transferNos.size() < pageSize)
|
||||
{
|
||||
percent = 99;
|
||||
}
|
||||
put(asyncTaskId, TabOrderCompareTask.STATUS_RUNNING, percent,
|
||||
"scanning page=" + pageNum + " scanned=" + scanned + " missing=" + missing,
|
||||
scanned, matched, missing, null, missingSample);
|
||||
if (transferNos.size() < pageSize)
|
||||
{
|
||||
break;
|
||||
}
|
||||
pageNum++;
|
||||
}
|
||||
Map<String, Object> resp = new LinkedHashMap<String, Object>();
|
||||
resp.put("asyncTaskId", asyncTaskId);
|
||||
resp.put("scanned", Integer.valueOf(scanned));
|
||||
resp.put("matched", Integer.valueOf(matched));
|
||||
resp.put("missing", Integer.valueOf(missing));
|
||||
resp.put("missingTransferNos", missingSample);
|
||||
resp.put("missingTruncated", Boolean.valueOf(missing > missingSample.size()));
|
||||
put(asyncTaskId, TabOrderCompareTask.STATUS_SUCCESS, 100,
|
||||
"done scanned=" + scanned + " missing=" + missing,
|
||||
scanned, matched, missing, null, missingSample);
|
||||
TabOrderSyncEsLog.save(billSearchStore, asyncTaskId, "COMPARE_DONE", query, resp, "SUCCESS", null);
|
||||
log.info("orderSource compare done asyncTaskId={} scanned={} matched={} missing={}",
|
||||
asyncTaskId, Integer.valueOf(scanned), Integer.valueOf(matched), Integer.valueOf(missing));
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.error("orderSource compare failed asyncTaskId={}", asyncTaskId, e);
|
||||
String msg = e.getMessage();
|
||||
if (StringUtils.isEmpty(msg))
|
||||
{
|
||||
msg = e.getClass().getSimpleName();
|
||||
}
|
||||
put(asyncTaskId, TabOrderCompareTask.STATUS_FAIL, 100, "failed", 0, 0, 0, msg, null);
|
||||
TabOrderSyncEsLog.save(billSearchStore, asyncTaskId, "COMPARE_DONE", query, null, "FAIL", msg);
|
||||
}
|
||||
}
|
||||
|
||||
private Set<String> findExisting(List<String> transferNos, List<String> customerCodes)
|
||||
{
|
||||
Set<String> found = new HashSet<String>();
|
||||
if (transferNos == null || transferNos.isEmpty())
|
||||
{
|
||||
return found;
|
||||
}
|
||||
List<String> hit = prcOrderBillMapper.selectExistingTransferNos(transferNos, cleanCodes(customerCodes));
|
||||
if (hit == null)
|
||||
{
|
||||
return found;
|
||||
}
|
||||
for (String s : hit)
|
||||
{
|
||||
if (StringUtils.isNotEmpty(s))
|
||||
{
|
||||
found.add(s.trim().toUpperCase(Locale.ROOT));
|
||||
}
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
private static List<String> cleanCodes(List<String> codes)
|
||||
{
|
||||
if (codes == null || codes.isEmpty())
|
||||
{
|
||||
return null;
|
||||
}
|
||||
List<String> out = new ArrayList<String>();
|
||||
for (String c : codes)
|
||||
{
|
||||
if (StringUtils.isNotEmpty(c))
|
||||
{
|
||||
out.add(c.trim());
|
||||
}
|
||||
}
|
||||
return out.isEmpty() ? null : out;
|
||||
}
|
||||
|
||||
public void put(String taskId, String status, int percent, String message,
|
||||
int scanned, int matched, int missing, String error, List<String> missingSample)
|
||||
{
|
||||
TabOrderCompareTask task = new TabOrderCompareTask();
|
||||
task.setTaskId(taskId);
|
||||
task.setStatus(status);
|
||||
task.setPercent(percent);
|
||||
task.setMessage(message);
|
||||
task.setScanned(scanned);
|
||||
task.setMatched(matched);
|
||||
task.setMissing(missing);
|
||||
task.setError(error);
|
||||
if (missingSample != null)
|
||||
{
|
||||
task.setMissingTransferNos(new ArrayList<String>(missingSample));
|
||||
}
|
||||
redisCache.setCacheObject(CACHE_KEY_PREFIX + taskId, task, TTL_MINUTES, TimeUnit.MINUTES);
|
||||
}
|
||||
|
||||
public TabOrderCompareTask get(String taskId)
|
||||
{
|
||||
return redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user