同步计费优化
This commit is contained in:
+3
-2
@@ -136,9 +136,10 @@ public class TabOrderMonitorController extends BaseController
|
||||
@PostMapping("/syncByFilter")
|
||||
public AjaxResult syncByFilter(@RequestBody TabOrderMonitorQuery query)
|
||||
{
|
||||
if (!hasRequiredFilters(query))
|
||||
// 与列表查询同口径:有单号即可同步;无单号时必须客户 + 提货时间
|
||||
if (!hasListFilters(query))
|
||||
{
|
||||
return error("customerCodes and pickup time required");
|
||||
return error("orderIdentifiers or (customer and pickup time) required");
|
||||
}
|
||||
customerKeywordResolver.resolve(query);
|
||||
// 含大批量单号时,异步任务启动时会自行分批收集,这里不预先 count,避免重复全量扫描
|
||||
|
||||
@@ -26,8 +26,10 @@ bms:
|
||||
# Max messages popped per tick; processed by consumer-threads pool
|
||||
consumer-batch-size: 32
|
||||
consumer-threads: 8
|
||||
# Concurrent threads for monitor order-source sync (2~16)
|
||||
order-source-sync-threads: 8
|
||||
# Concurrent threads for monitor order-source sync (2~16).
|
||||
# 同步已直接复用订单行(跳过每单 Oracle 公网查询),瓶颈在 MySQL 计费写入,
|
||||
# 同客户行锁冲突由 billWithDbRetry 3 次重试兜底,16 路实测吞吐最高。
|
||||
order-source-sync-threads: 16
|
||||
oracle:
|
||||
# SLAVE datasource Oracle main-order table (schema.table allowed)
|
||||
order-table: WEISHI_USER.TAB_ORDER
|
||||
|
||||
+78
-19
@@ -43,6 +43,12 @@ public class TabOrderMonitorService
|
||||
/** Oracle 单个 IN 列表上限 1000,分批收集时每批取 900 留安全余量。 */
|
||||
public static final int IDENTIFIER_BATCH_SIZE = 900;
|
||||
|
||||
/**
|
||||
* 服务端防 OOM 硬上限:正常入口前端限 1 万单号;此处仅防止误粘贴/调用方传入
|
||||
* 海量单号导致收集结果(每单约 ID+提货时间两个字段)撑爆堆内存。
|
||||
*/
|
||||
public static final int IDENTIFIER_HARD_LIMIT = 200000;
|
||||
|
||||
/**
|
||||
* 分批收集并行度:Oracle 从库连接池 maxActive=20,6 路并行与同步任务并发合计仍有余量。
|
||||
* 实测相同 12000 单号:串行约 18s(冷),6 路并行约 1s。
|
||||
@@ -296,7 +302,7 @@ public class TabOrderMonitorService
|
||||
return doCollectMatchedKeys(query, resolveTable());
|
||||
}
|
||||
|
||||
/** 同步任务按已收集好的主键切片拉取明细,顺序与 keys 切片一致。 */
|
||||
/** 同步任务按已收集好的主键切片拉取明细(VO),顺序与 keys 切片一致。 */
|
||||
@DataSource(DataSourceType.SLAVE)
|
||||
public List<TabOrderMonitorVo> selectPageByKeys(List<MatchedOrderKey> keySlice)
|
||||
{
|
||||
@@ -304,10 +310,38 @@ public class TabOrderMonitorService
|
||||
{
|
||||
return Collections.emptyList();
|
||||
}
|
||||
return doPageByKeys(resolveTable(), keySlice, 1, keySlice.size());
|
||||
return toVoList(selectRowsByKeys(keySlice));
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步任务按已收集好的主键切片拉取全行数据(VO + Oracle 原始行),
|
||||
* 原始行直接用于计费映射,省去每单再次公网查询 Oracle。
|
||||
*/
|
||||
@DataSource(DataSourceType.SLAVE)
|
||||
public List<TabOrderRow> selectRowsByKeys(List<MatchedOrderKey> keySlice)
|
||||
{
|
||||
if (keySlice == null || keySlice.isEmpty())
|
||||
{
|
||||
return Collections.emptyList();
|
||||
}
|
||||
return doRowsByKeys(resolveTable(), keySlice, 1, keySlice.size());
|
||||
}
|
||||
|
||||
/** 非单号收集路径(客户+时间等条件)同步任务用:按页拉全行数据。 */
|
||||
@DataSource(DataSourceType.SLAVE)
|
||||
public List<TabOrderRow> selectRownumRows(TabOrderMonitorQuery query, int pageNum, int pageSize)
|
||||
{
|
||||
validate(query);
|
||||
int[] page = normalizePage(pageNum, pageSize);
|
||||
return doSelectRownumRows(resolveTable(), query, page[0], page[1]);
|
||||
}
|
||||
|
||||
private List<TabOrderMonitorVo> doSelectRownumPage(String table, TabOrderMonitorQuery query, int pn, int ps)
|
||||
{
|
||||
return toVoList(doSelectRownumRows(table, query, pn, ps));
|
||||
}
|
||||
|
||||
private List<TabOrderRow> doSelectRownumRows(String table, TabOrderMonitorQuery query, int pn, int ps)
|
||||
{
|
||||
long endRow = (long) pn * ps;
|
||||
long startRow = endRow - ps;
|
||||
@@ -320,7 +354,7 @@ public class TabOrderMonitorService
|
||||
query.getCustomerCodes() == null ? 0 : query.getCustomerCodes().size(),
|
||||
query.getBeginTime(), query.getEndTime(), pn, ps,
|
||||
rows == null ? 0 : rows.size(), System.currentTimeMillis() - t0);
|
||||
return toVoList(rows);
|
||||
return wrapRows(rows);
|
||||
}
|
||||
catch (RuntimeException e)
|
||||
{
|
||||
@@ -332,6 +366,34 @@ public class TabOrderMonitorService
|
||||
}
|
||||
}
|
||||
|
||||
private static List<TabOrderRow> wrapRows(List<Map<String, Object>> rows)
|
||||
{
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<TabOrderRow> list = new ArrayList<TabOrderRow>(rows.size());
|
||||
for (Map<String, Object> row : rows)
|
||||
{
|
||||
list.add(new TabOrderRow(toVo(row), row));
|
||||
}
|
||||
return list;
|
||||
}
|
||||
|
||||
private static List<TabOrderMonitorVo> toVoList(List<TabOrderRow> rows)
|
||||
{
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<TabOrderMonitorVo> list = new ArrayList<TabOrderMonitorVo>(rows.size());
|
||||
for (TabOrderRow row : rows)
|
||||
{
|
||||
list.add(row.getVo());
|
||||
}
|
||||
return list;
|
||||
}
|
||||
|
||||
/**
|
||||
* 核心:单号按 {@link #IDENTIFIER_BATCH_SIZE} 分批,{@link #COLLECT_PARALLELISM} 路并行
|
||||
* 查 Oracle(每批 UNION ALL 三列各走索引),Java 侧按 ID 去重,
|
||||
@@ -344,6 +406,11 @@ public class TabOrderMonitorService
|
||||
{
|
||||
return Collections.emptyList();
|
||||
}
|
||||
if (identifiers.size() > IDENTIFIER_HARD_LIMIT)
|
||||
{
|
||||
throw new ServiceException("单号数量超过服务端上限 " + IDENTIFIER_HARD_LIMIT
|
||||
+ ",请分批查询/同步,避免内存溢出");
|
||||
}
|
||||
String fingerprint = collectFingerprint(table, query);
|
||||
long now = System.currentTimeMillis();
|
||||
CacheEntry cached = collectCache.get(fingerprint);
|
||||
@@ -485,8 +552,14 @@ public class TabOrderMonitorService
|
||||
return s == null ? "" : s;
|
||||
}
|
||||
|
||||
/** 已排序主键的内存分页切片,并按切片顺序回查明细。 */
|
||||
/** 已排序主键的内存分页切片,并按切片顺序回查明细(VO)。 */
|
||||
private List<TabOrderMonitorVo> doPageByKeys(String table, List<MatchedOrderKey> orderedKeys, int pn, int ps)
|
||||
{
|
||||
return toVoList(doRowsByKeys(table, orderedKeys, pn, ps));
|
||||
}
|
||||
|
||||
/** 已排序主键的内存分页切片,按切片顺序回查全行(VO + Oracle 原始行)。 */
|
||||
private List<TabOrderRow> doRowsByKeys(String table, List<MatchedOrderKey> orderedKeys, int pn, int ps)
|
||||
{
|
||||
if (orderedKeys == null || orderedKeys.isEmpty())
|
||||
{
|
||||
@@ -530,7 +603,7 @@ public class TabOrderMonitorService
|
||||
orderedRows.add(row);
|
||||
}
|
||||
}
|
||||
return toVoList(orderedRows);
|
||||
return wrapRows(orderedRows);
|
||||
}
|
||||
catch (RuntimeException e)
|
||||
{
|
||||
@@ -558,20 +631,6 @@ public class TabOrderMonitorService
|
||||
return query != null && query.getOrderIdentifiers() != null && !query.getOrderIdentifiers().isEmpty();
|
||||
}
|
||||
|
||||
private static List<TabOrderMonitorVo> toVoList(List<Map<String, Object>> rows)
|
||||
{
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<TabOrderMonitorVo> list = new ArrayList<TabOrderMonitorVo>(rows.size());
|
||||
for (Map<String, Object> row : rows)
|
||||
{
|
||||
list.add(toVo(row));
|
||||
}
|
||||
return list;
|
||||
}
|
||||
|
||||
private void validate(TabOrderMonitorQuery query)
|
||||
{
|
||||
if (query == null)
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
package com.bms.price.oracle;
|
||||
|
||||
import java.util.Map;
|
||||
import com.bms.price.domain.vo.TabOrderMonitorVo;
|
||||
|
||||
/**
|
||||
* 订单列表/同步使用的一行数据:列表展示用 VO,同步计费用 Oracle 原始全行 Map,
|
||||
* 避免同步时按单号再次公网查询 Oracle(findMainOrder)。
|
||||
*/
|
||||
public class TabOrderRow
|
||||
{
|
||||
private final TabOrderMonitorVo vo;
|
||||
private final Map<String, Object> raw;
|
||||
|
||||
public TabOrderRow(TabOrderMonitorVo vo, Map<String, Object> raw)
|
||||
{
|
||||
this.vo = vo;
|
||||
this.raw = raw;
|
||||
}
|
||||
|
||||
public TabOrderMonitorVo getVo()
|
||||
{
|
||||
return vo;
|
||||
}
|
||||
|
||||
public Map<String, Object> getRaw()
|
||||
{
|
||||
return raw;
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package com.bms.price.service;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import com.bms.price.domain.PrcBillingTask;
|
||||
import com.bms.price.domain.PrcBillingTaskRetryTask;
|
||||
import com.bms.price.domain.vo.BillingTaskRetryByFilterRequest;
|
||||
@@ -14,6 +15,12 @@ public interface IPrcBillingTaskService
|
||||
/** Monitor sync: create task then process immediately (no extra MQ). */
|
||||
PrcBillingTask submitAndProcess(BillingTaskSubmitRequest request);
|
||||
|
||||
/**
|
||||
* Monitor sync with upstream Oracle full row already fetched by order list:
|
||||
* billing maps the row directly, skipping per-order Oracle public-network queries.
|
||||
*/
|
||||
PrcBillingTask submitAndProcess(BillingTaskSubmitRequest request, Map<String, Object> upstreamRow);
|
||||
|
||||
/** MQ consumer entry: Oracle query -> bill -> writeback success status. */
|
||||
void processTask(Long taskId);
|
||||
|
||||
|
||||
+102
-51
@@ -115,6 +115,47 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
|
||||
return billingTaskMapper.selectPrcBillingTaskById(task.getTaskId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public PrcBillingTask submitAndProcess(BillingTaskSubmitRequest request, Map<String, Object> upstreamRow)
|
||||
{
|
||||
PrcBillingTask task = createWaitingTask(request, "monitor");
|
||||
// 退款不依赖上游行;行缺失时也回退到按单号查询 Oracle 的原有路径
|
||||
if (upstreamRow == null || upstreamRow.isEmpty()
|
||||
|| BillTaskType.REFUND == BillTaskType.fromCode(task.getBillType()))
|
||||
{
|
||||
processTask(task.getTaskId());
|
||||
}
|
||||
else
|
||||
{
|
||||
processTaskWithUpstreamRow(task, upstreamRow);
|
||||
}
|
||||
return billingTaskMapper.selectPrcBillingTaskById(task.getTaskId());
|
||||
}
|
||||
|
||||
/** 订单列表已拉回 Oracle 全行时直接计费,跳过 findMainOrder 的多次公网往返。 */
|
||||
private void processTaskWithUpstreamRow(PrcBillingTask task, Map<String, Object> row)
|
||||
{
|
||||
Long taskId = task.getTaskId();
|
||||
if (BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus()))
|
||||
{
|
||||
log.info("billing task already success taskId={}", taskId);
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
if (task.getBillTime() == null)
|
||||
{
|
||||
throw new ServiceException("task billTime empty (billing time from mobile required)");
|
||||
}
|
||||
billFromUpstreamRow(task, row);
|
||||
}
|
||||
catch (RuntimeException e)
|
||||
{
|
||||
log.error("billing task failed taskId={}", taskId, e);
|
||||
writeback(task, BillTaskSuccessStatus.FAILED, null, cut(e.getMessage(), 900), null);
|
||||
}
|
||||
}
|
||||
|
||||
private PrcBillingTask createWaitingTask(BillingTaskSubmitRequest request, String createBy)
|
||||
{
|
||||
if (request == null)
|
||||
@@ -198,57 +239,7 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
|
||||
{
|
||||
throw new ServiceException("Oracle main order not found");
|
||||
}
|
||||
TmsPriceBillRequest billReq = oracleOrderBillMapper.toBillRequest(row,
|
||||
billingProperties.getOracle().getDefaultProductCode());
|
||||
if (billReq == null || StringUtils.isEmpty(billReq.getOrderNo()))
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped orderNo empty");
|
||||
}
|
||||
if (StringUtils.isEmpty(billReq.getCustomerCode()))
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped customerCode empty");
|
||||
}
|
||||
if (StringUtils.isEmpty(billReq.getProductCode()))
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped productCode empty");
|
||||
}
|
||||
if (billReq.getWeight() == null || billReq.getWeight().signum() <= 0)
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped weight invalid");
|
||||
}
|
||||
// Identity keys from mobile inbound are source of truth; do not replace with Oracle remap.
|
||||
if (StringUtils.isNotEmpty(task.getTransferNo()))
|
||||
{
|
||||
billReq.setTransferNo(task.getTransferNo());
|
||||
}
|
||||
if (StringUtils.isNotEmpty(task.getTrackingNo()))
|
||||
{
|
||||
billReq.setTrackingNo(task.getTrackingNo());
|
||||
}
|
||||
if (StringUtils.isNotEmpty(task.getOrderNo()))
|
||||
{
|
||||
billReq.setOrderNo(task.getOrderNo());
|
||||
}
|
||||
// Billing datetime from task.bill_time (keep HH:mm:ss), not Oracle CREATE_DATE.
|
||||
billReq.setBizDate(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(task.getBillTime()));
|
||||
task.setCustomerCode(billReq.getCustomerCode());
|
||||
|
||||
TmsPriceQuoteResult result = billWithDbRetry(taskId, billReq);
|
||||
// Soft-fail (product/country/weight miss): order bill already persisted with zero amounts;
|
||||
// still mark task FAILED so it appears in abnormal-task list for retry/ops.
|
||||
if (result != null && BillBillingFailStatus.isFail(result.getBillingFailStatus()))
|
||||
{
|
||||
String err = StringUtils.isNotEmpty(result.getMatchMessage())
|
||||
? result.getMatchMessage()
|
||||
: ("billingFailStatus=" + result.getBillingFailStatus());
|
||||
writeback(task, BillTaskSuccessStatus.FAILED, result.getBillId(), cut(err, 900),
|
||||
BillBillingFailStatus.toTaskFailCode(result.getBillingFailStatus()));
|
||||
}
|
||||
else
|
||||
{
|
||||
writeback(task, BillTaskSuccessStatus.SUCCESS, result == null ? null : result.getBillId(),
|
||||
null, null);
|
||||
}
|
||||
billFromUpstreamRow(task, row);
|
||||
}
|
||||
catch (RuntimeException e)
|
||||
{
|
||||
@@ -257,6 +248,66 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 用上游 Oracle 全行数据映射并计费(公共逻辑:MQ 处理与订单列表同步复用)。
|
||||
* 订单列表同步时行数据已由分页查询拉回,避免每单再走 findMainOrder 的公网往返。
|
||||
*/
|
||||
private void billFromUpstreamRow(PrcBillingTask task, Map<String, Object> row)
|
||||
{
|
||||
Long taskId = task.getTaskId();
|
||||
TmsPriceBillRequest billReq = oracleOrderBillMapper.toBillRequest(row,
|
||||
billingProperties.getOracle().getDefaultProductCode());
|
||||
if (billReq == null || StringUtils.isEmpty(billReq.getOrderNo()))
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped orderNo empty");
|
||||
}
|
||||
if (StringUtils.isEmpty(billReq.getCustomerCode()))
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped customerCode empty");
|
||||
}
|
||||
if (StringUtils.isEmpty(billReq.getProductCode()))
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped productCode empty");
|
||||
}
|
||||
if (billReq.getWeight() == null || billReq.getWeight().signum() <= 0)
|
||||
{
|
||||
throw new ServiceException("Oracle order mapped weight invalid");
|
||||
}
|
||||
// Identity keys from mobile inbound are source of truth; do not replace with Oracle remap.
|
||||
if (StringUtils.isNotEmpty(task.getTransferNo()))
|
||||
{
|
||||
billReq.setTransferNo(task.getTransferNo());
|
||||
}
|
||||
if (StringUtils.isNotEmpty(task.getTrackingNo()))
|
||||
{
|
||||
billReq.setTrackingNo(task.getTrackingNo());
|
||||
}
|
||||
if (StringUtils.isNotEmpty(task.getOrderNo()))
|
||||
{
|
||||
billReq.setOrderNo(task.getOrderNo());
|
||||
}
|
||||
// Billing datetime from task.bill_time (keep HH:mm:ss), not Oracle CREATE_DATE.
|
||||
billReq.setBizDate(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(task.getBillTime()));
|
||||
task.setCustomerCode(billReq.getCustomerCode());
|
||||
|
||||
TmsPriceQuoteResult result = billWithDbRetry(taskId, billReq);
|
||||
// Soft-fail (product/country/weight miss): order bill already persisted with zero amounts;
|
||||
// still mark task FAILED so it appears in abnormal-task list for retry/ops.
|
||||
if (result != null && BillBillingFailStatus.isFail(result.getBillingFailStatus()))
|
||||
{
|
||||
String err = StringUtils.isNotEmpty(result.getMatchMessage())
|
||||
? result.getMatchMessage()
|
||||
: ("billingFailStatus=" + result.getBillingFailStatus());
|
||||
writeback(task, BillTaskSuccessStatus.FAILED, result.getBillId(), cut(err, 900),
|
||||
BillBillingFailStatus.toTaskFailCode(result.getBillingFailStatus()));
|
||||
}
|
||||
else
|
||||
{
|
||||
writeback(task, BillTaskSuccessStatus.SUCCESS, result == null ? null : result.getBillId(),
|
||||
null, null);
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolve orderNo for REFUND from task keys or existing bill (no Oracle quote required). */
|
||||
private String resolveRefundOrderNo(PrcBillingTask task)
|
||||
{
|
||||
|
||||
@@ -43,11 +43,14 @@ public class TabOrderSyncAsyncRunner
|
||||
|
||||
public static final String CACHE_KEY_PREFIX = "billing:orderSource:sync:";
|
||||
public static final int TTL_MINUTES = 120;
|
||||
public static final int DEFAULT_BATCH = 50;
|
||||
/** 每页拉取的订单行数:行内直接带 Oracle 全行数据,单页大些可减少分页往返。 */
|
||||
public static final int DEFAULT_BATCH = 100;
|
||||
public static final int MIN_BATCH = 20;
|
||||
public static final int MAX_BATCH = 200;
|
||||
public static final int MIN_THREADS = 2;
|
||||
public static final int MAX_THREADS = 16;
|
||||
/** 进度写入 Redis 的最小间隔(条数),避免每单一次 Redis SET。 */
|
||||
private static final int PROGRESS_REPORT_EVERY = 10;
|
||||
|
||||
@Autowired
|
||||
private TabOrderMonitorService tabOrderMonitorService;
|
||||
@@ -97,7 +100,7 @@ public class TabOrderSyncAsyncRunner
|
||||
int pageNum = 1;
|
||||
while (true)
|
||||
{
|
||||
List<TabOrderMonitorVo> rows;
|
||||
List<com.bms.price.oracle.TabOrderRow> rows;
|
||||
if (keys != null)
|
||||
{
|
||||
int from = (pageNum - 1) * batch;
|
||||
@@ -105,12 +108,12 @@ public class TabOrderSyncAsyncRunner
|
||||
{
|
||||
break;
|
||||
}
|
||||
rows = tabOrderMonitorService.selectPageByKeys(
|
||||
rows = tabOrderMonitorService.selectRowsByKeys(
|
||||
keys.subList(from, Math.min(from + batch, keys.size())));
|
||||
}
|
||||
else
|
||||
{
|
||||
rows = tabOrderMonitorService.selectPage(query, pageNum, batch);
|
||||
rows = tabOrderMonitorService.selectRownumRows(query, pageNum, batch);
|
||||
}
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
@@ -122,9 +125,9 @@ public class TabOrderSyncAsyncRunner
|
||||
total, processed.get(), ok.get(), fail.get(), null);
|
||||
|
||||
List<Callable<Boolean>> jobs = new ArrayList<Callable<Boolean>>(rows.size());
|
||||
for (TabOrderMonitorVo row : rows)
|
||||
for (com.bms.price.oracle.TabOrderRow row : rows)
|
||||
{
|
||||
final TabOrderMonitorVo one = row;
|
||||
final com.bms.price.oracle.TabOrderRow one = row;
|
||||
jobs.add(() -> {
|
||||
boolean success = syncOne(one, failDetails);
|
||||
if (success)
|
||||
@@ -136,10 +139,14 @@ public class TabOrderSyncAsyncRunner
|
||||
fail.incrementAndGet();
|
||||
}
|
||||
int done = processed.incrementAndGet();
|
||||
int percent = Math.min(99, (int) ((done * 100L) / Math.max(1, total)));
|
||||
put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, percent,
|
||||
"processing " + done + "/" + total,
|
||||
total, done, ok.get(), fail.get(), null);
|
||||
// 降频写 Redis(每 10 条或最后一条),轮询端 1.5s 一次无需更细粒度
|
||||
if (done % PROGRESS_REPORT_EVERY == 0 || done >= total)
|
||||
{
|
||||
int percent = Math.min(99, (int) ((done * 100L) / Math.max(1, total)));
|
||||
put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, percent,
|
||||
"processing " + done + "/" + total,
|
||||
total, done, ok.get(), fail.get(), null);
|
||||
}
|
||||
return Boolean.valueOf(success);
|
||||
});
|
||||
}
|
||||
@@ -196,6 +203,25 @@ public class TabOrderSyncAsyncRunner
|
||||
* @param billType optional PENDING/REBILL/REFUND; null = auto PENDING/REBILL
|
||||
*/
|
||||
public Map<String, Object> syncRowsParallel(List<TabOrderMonitorVo> rows, String billType)
|
||||
{
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
return syncRowsParallelRows(java.util.Collections.<com.bms.price.oracle.TabOrderRow>emptyList(), billType);
|
||||
}
|
||||
List<com.bms.price.oracle.TabOrderRow> wrapped =
|
||||
new ArrayList<com.bms.price.oracle.TabOrderRow>(rows.size());
|
||||
for (TabOrderMonitorVo vo : rows)
|
||||
{
|
||||
wrapped.add(new com.bms.price.oracle.TabOrderRow(vo, null));
|
||||
}
|
||||
return syncRowsParallelRows(wrapped, billType);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sync or refund selected rows carrying the upstream Oracle full row (skip re-query).
|
||||
* @param billType optional PENDING/REBILL/REFUND; null = auto PENDING/REBILL
|
||||
*/
|
||||
public Map<String, Object> syncRowsParallelRows(List<com.bms.price.oracle.TabOrderRow> rows, String billType)
|
||||
{
|
||||
Map<String, Object> data = new HashMap<String, Object>();
|
||||
List<Map<String, Object>> details = new ArrayList<Map<String, Object>>();
|
||||
@@ -215,16 +241,18 @@ public class TabOrderSyncAsyncRunner
|
||||
try
|
||||
{
|
||||
List<Callable<Map<String, Object>>> jobs = new ArrayList<Callable<Map<String, Object>>>(rows.size());
|
||||
for (TabOrderMonitorVo row : rows)
|
||||
for (com.bms.price.oracle.TabOrderRow row : rows)
|
||||
{
|
||||
final TabOrderMonitorVo one = row;
|
||||
final com.bms.price.oracle.TabOrderRow one = row;
|
||||
jobs.add(() -> {
|
||||
Map<String, Object> item = new HashMap<String, Object>();
|
||||
item.put("orderId", one == null ? null : one.getOrderId());
|
||||
item.put("transferOrderId", one == null ? null : one.getTransferOrderId());
|
||||
TabOrderMonitorVo vo = one == null ? null : one.getVo();
|
||||
item.put("orderId", vo == null ? null : vo.getOrderId());
|
||||
item.put("transferOrderId", vo == null ? null : vo.getTransferOrderId());
|
||||
try
|
||||
{
|
||||
PrcBillingTask task = billingTaskService.submitAndProcess(toSubmit(one, fixedBillType));
|
||||
PrcBillingTask task = billingTaskService.submitAndProcess(
|
||||
toSubmit(vo, fixedBillType), one == null ? null : one.getRaw());
|
||||
item.put("successStatus", task == null ? null : task.getSuccessStatus());
|
||||
item.put("errorMessage", task == null ? null : task.getErrorMessage());
|
||||
item.put("billType", task == null ? fixedBillType : task.getBillType());
|
||||
@@ -280,26 +308,28 @@ public class TabOrderSyncAsyncRunner
|
||||
return data;
|
||||
}
|
||||
|
||||
private boolean syncOne(TabOrderMonitorVo row, List<Map<String, Object>> failDetails)
|
||||
private boolean syncOne(com.bms.price.oracle.TabOrderRow row, List<Map<String, Object>> failDetails)
|
||||
{
|
||||
TabOrderMonitorVo vo = row == null ? null : row.getVo();
|
||||
try
|
||||
{
|
||||
PrcBillingTask task = billingTaskService.submitAndProcess(toSubmit(row));
|
||||
PrcBillingTask task = billingTaskService.submitAndProcess(
|
||||
toSubmit(vo), row == null ? null : row.getRaw());
|
||||
boolean success = task != null
|
||||
&& BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus());
|
||||
if (!success)
|
||||
{
|
||||
collectFailDetail(failDetails, row, task, null);
|
||||
collectFailDetail(failDetails, vo, task, null);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
catch (RuntimeException e)
|
||||
{
|
||||
log.warn("orderSource sync failed orderId={} transfer={} msg={}",
|
||||
row == null ? null : row.getOrderId(),
|
||||
row == null ? null : row.getTransferOrderId(),
|
||||
vo == null ? null : vo.getOrderId(),
|
||||
vo == null ? null : vo.getTransferOrderId(),
|
||||
e.getMessage());
|
||||
collectFailDetail(failDetails, row, null, e.getMessage());
|
||||
collectFailDetail(failDetails, vo, null, e.getMessage());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -446,7 +476,7 @@ public class TabOrderSyncAsyncRunner
|
||||
|
||||
public static int clampThreads(Integer threads)
|
||||
{
|
||||
int n = threads == null || threads.intValue() <= 0 ? 8 : threads.intValue();
|
||||
int n = threads == null || threads.intValue() <= 0 ? 16 : threads.intValue();
|
||||
if (n < MIN_THREADS)
|
||||
{
|
||||
return MIN_THREADS;
|
||||
|
||||
@@ -131,8 +131,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
<select id="selectTabOrderMonitorPage" resultType="java.util.HashMap">
|
||||
SELECT * FROM (
|
||||
SELECT inner_q.*, ROWNUM AS rn FROM (
|
||||
SELECT
|
||||
<include refid="tabOrderMonitorColumns"/>
|
||||
<!-- 同步计费需要全行数据(toBillRequest 读取数十列),故取 o.* 而非展示列 -->
|
||||
SELECT o.*
|
||||
FROM ${orderTable} o
|
||||
WHERE
|
||||
<include refid="tabOrderMonitorWhere"/>
|
||||
@@ -211,8 +211,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
|
||||
<!-- 内存分页切片后的主键回查(一批最多一个页大小,远低于 IN 上限) -->
|
||||
<select id="selectTabOrderMonitorByIds" resultType="java.util.HashMap">
|
||||
SELECT
|
||||
<include refid="tabOrderMonitorColumns"/>
|
||||
<!-- 全行数据:同步计费时直接复用,不再按单号回查 Oracle -->
|
||||
SELECT o.*
|
||||
FROM ${orderTable} o
|
||||
WHERE o.ID IN
|
||||
<foreach collection="ids" item="mid" open="(" separator="," close=")">
|
||||
|
||||
@@ -182,8 +182,12 @@ function submit(skipConfirm) {
|
||||
const q = query.value || {}
|
||||
const codes = Array.isArray(q.customerCodes) ? q.customerCodes.filter(v => v != null && String(v).trim() !== '') : []
|
||||
const keywords = Array.isArray(q.customerKeywords) ? q.customerKeywords.filter(v => v != null && String(v).trim() !== '') : []
|
||||
if ((!codes.length && !keywords.length) || !q.beginTime || !q.endTime) {
|
||||
proxy.$modal.msgError('请先输入客户并选择提货时间')
|
||||
const identifiers = Array.isArray(q.orderIdentifiers)
|
||||
? q.orderIdentifiers.filter(v => v != null && String(v).trim() !== '')
|
||||
: []
|
||||
// 与查询同口径:有单号即可同步;无单号时必须客户 + 提货时间
|
||||
if (!identifiers.length && ((!codes.length && !keywords.length) || !q.beginTime || !q.endTime)) {
|
||||
proxy.$modal.msgError('请输入单号,或先输入客户并选择提货时间')
|
||||
return
|
||||
}
|
||||
const start = () => {
|
||||
@@ -209,7 +213,7 @@ function submit(skipConfirm) {
|
||||
orderId: q.orderId || undefined,
|
||||
orgOrderId: q.orgOrderId || undefined,
|
||||
transferOrderId: q.transferOrderId || undefined,
|
||||
batchSize: 50
|
||||
batchSize: 100
|
||||
}).then(res => {
|
||||
const asyncTaskId = res.data?.asyncTaskId || res.data
|
||||
if (!asyncTaskId) {
|
||||
|
||||
@@ -131,7 +131,7 @@ const searchColumns = [
|
||||
label: '单号',
|
||||
prop: 'orderIdentifiers',
|
||||
component: 'textarea',
|
||||
placeholder: '订单号/运单号/转单号,一行一个,支持数万个',
|
||||
placeholder: '订单号/运单号/转单号,一行一个,单次最多 1 万个',
|
||||
props: {
|
||||
rows: 2,
|
||||
autosize: { minRows: 1, maxRows: 4 },
|
||||
@@ -215,8 +215,8 @@ function rowKey(row) {
|
||||
return (row && (row.orderId || row.transferOrderId || row.orgOrderId)) || ''
|
||||
}
|
||||
|
||||
/** 单次查询可接受的单号上限(后端按 900/批分批走索引,过高会明显增加等待时间) */
|
||||
const MAX_ORDER_IDENTIFIERS = 50000
|
||||
/** 单次查询/同步可接受的单号上限(后端按 900/批分批走索引,另有服务端 20 万防 OOM 硬兜底) */
|
||||
const MAX_ORDER_IDENTIFIERS = 10000
|
||||
/** 产品检测走单条 SQL(Oracle IN 上限 1000),与后端保持一致 */
|
||||
const MAX_PRODUCT_CHECK_IDENTIFIERS = 1000
|
||||
|
||||
@@ -265,11 +265,23 @@ function queryList(params) {
|
||||
|
||||
function openFilterSync() {
|
||||
const q = buildFilterQuery()
|
||||
if (!(q.customerKeywords && q.customerKeywords.length) || !q.beginTime || !q.endTime) {
|
||||
ElMessage.warning('按条件同步计费需先输入客户并选择提货时间')
|
||||
const identifierCount = (q.orderIdentifiers || []).length
|
||||
const hasIdentifier = identifierCount > 0
|
||||
const hasCustomer = !!(q.customerKeywords && q.customerKeywords.length)
|
||||
const hasTime = !!q.beginTime && !!q.endTime
|
||||
// 与查询同口径:有单号即可同步;无单号时必须客户+提货时间
|
||||
if (!hasIdentifier && !(hasCustomer && hasTime)) {
|
||||
ElMessage.warning('请输入单号,或选择客户和提货时间后再同步计费')
|
||||
return
|
||||
}
|
||||
proxy.$modal.confirm('确认按当前筛选分页查询并异步同步计费?', '二次确认').then(() => {
|
||||
if (identifierCount > MAX_ORDER_IDENTIFIERS) {
|
||||
ElMessage.warning(`单次最多同步 ${MAX_ORDER_IDENTIFIERS} 个单号,当前 ${identifierCount} 个,请减少后再同步`)
|
||||
return
|
||||
}
|
||||
const tip = hasIdentifier && !(hasCustomer && hasTime)
|
||||
? `将按 ${identifierCount} 个单号查询匹配订单并异步同步计费,确认继续?`
|
||||
: '确认按当前筛选分页查询并异步同步计费?'
|
||||
proxy.$modal.confirm(tip, '二次确认').then(() => {
|
||||
syncDialogRef.value?.open(q, true)
|
||||
}).catch(() => {})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user