wuweihua/BMS
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of https://git.lionquexpress.com/wuweihua/BMS

Browse Source
This commit is contained in:
wuweihua
2026-10-16 11:27:48 +08:00
parent 2431a5caac 760e3807f3
commit b3f79ffc4c
14 changed files with 346 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bms-back/bms-admin/src/main/java/com/bms/web/controller/ar/ArBillController.java
+5 -4
View File
@@ -6,6 +6,7 @@ import java.util.Date;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import jakarta.validation.Valid;
import com.github.pagehelper.PageHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.WebDataBinder;
@@ -120,10 +121,10 @@ public class ArBillController extends BaseController {
* 查询生成账单前的订单计费结果。
*/
@PreAuthorize("@ss.hasPermi('ar:bill:generate')")
@GetMapping("/generate/preview")
public AjaxResult preview(ArBillGenerateQuery query, HttpServletRequest request) {
rejectMultiValueParameter(request, "customerId");
startPage();
@PostMapping("/generate/preview")
public AjaxResult preview(@RequestBody ArBillGenerateQuery query) {
PageHelper.startPage(query == null || query.getPageNum() == null ? 1 : query.getPageNum(),
query == null || query.getPageSize() == null ? 10 : query.getPageSize()).setReasonable(true);
List<PrcOrderBill> list = billService.previewBill(query);
TableDataInfo pageData = getDataTable(list);
AjaxResult result = success();
bms-back/bms-admin/src/main/java/com/bms/web/controller/monitor/TabOrderMonitorController.java
+3 -2
View File
@@ -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，避免重复全量扫描
bms-back/bms-admin/src/main/resources/application.yml
+4 -2
View File
@@ -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
bms-back/bms-system/src/main/java/com/bms/ar/domain/vo/ArBillGenerateQuery.java
+9
View File
@@ -1,6 +1,9 @@
package com.bms.ar.domain.vo;
import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.bms.common.config.serializer.StrictDateDeserializer;
import org.springframework.format.annotation.DateTimeFormat;
import lombok.Data;
@@ -8,11 +11,17 @@ import lombok.Data;
@Data
public class ArBillGenerateQuery
{
private Integer pageNum;
private Integer pageSize;
private Long customerId;
private String customerCode;
@DateTimeFormat(pattern = "yyyy-MM-dd")
@JsonFormat(pattern = "yyyy-MM-dd")
@JsonDeserialize(using = StrictDateDeserializer.class)
private Date billPeriodEnd;
private String currency;
bms-back/bms-system/src/main/java/com/bms/ar/service/impl/ArBillServiceImpl.java
+21 -1
View File
@@ -135,7 +135,10 @@ public class ArBillServiceImpl implements IArBillService {
private static final int EXPLICIT_SOURCE_ID_LIMIT = 10000;
private static final int EXCLUDED_SOURCE_ID_LIMIT = 10000;
/** 单次生成查询最多接收的订单号、运单号或转单号数量。 */
private static final int ORDER_IDENTIFIER_LIMIT = 2000;
private static final int ORDER_IDENTIFIER_LIMIT = 20000;
/** 订单号会分别匹配订单号、运单号和转单号列；预留其他 SQL 参数空间后控制 MySQL 参数总数。 */
private static final int ORDER_IDENTIFIER_SQL_BIND_COUNT = 3;
private static final int MAX_SQL_PARAMETER_COUNT = 65000;
/**
* 模板明细行使用的边框、字体和换行样式；样式定义来自 ar_bill_template.xlsx。
*/
@@ -351,6 +354,9 @@ public class ArBillServiceImpl implements IArBillService {
throw new ServiceException("全选模式最多排除 " + EXCLUDED_SOURCE_ID_LIMIT
+ " 条订单，请缩小查询条件后再生成");
}
validateOrderIdentifierParameterCount(
normalizeOrderIdentifiers(request.getOrderOrTrackingNo()).size(),
allMatched ? excludedSourceBillIds.size() : sourceBillIds.size());
request.setSelectionMode(selectionMode);
String currency = StringUtils.isEmpty(request.getCurrency()) ? "CNY" : request.getCurrency().trim();
@@ -1990,6 +1996,12 @@ public class ArBillServiceImpl implements IArBillService {
|| query.getBillPeriodEnd() == null) {
throw new ServiceException("客户和账单截止日不能为空");
}
int identifierCount = normalizeOrderIdentifiers(query.getOrderOrTrackingNo()).size()
+ normalizeOrderIdentifiers(query.getExcludeOrderOrTrackingNo()).size();
if (identifierCount > ORDER_IDENTIFIER_LIMIT) {
throw new ServiceException("筛选单号和排除单号合计最多输入 " + ORDER_IDENTIFIER_LIMIT + " 个");
}
validateOrderIdentifierParameterCount(identifierCount, 0);
}
/** 可参与本次应收生成的订单状态；PARTIAL 只生成其中仍为 UNBILLED 的费用。 */
@@ -2041,6 +2053,14 @@ public class ArBillServiceImpl implements IArBillService {
return new ArrayList<>(uniqueIdentifiers);
}
/** 校验重复用于三列精确匹配的单号参数及选择/排除订单 ID 不会超过数据库参数预算。 */
private void validateOrderIdentifierParameterCount(int orderIdentifierCount, int billIdCount) {
long parameterCount = (long) orderIdentifierCount * ORDER_IDENTIFIER_SQL_BIND_COUNT + billIdCount;
if (parameterCount > MAX_SQL_PARAMETER_COUNT) {
throw new ServiceException("单号筛选与订单选择/排除条件过多，请减少单号或选择/排除订单数量后重试");
}
}
private long elapsedMillis(long startedAt) {
return (System.nanoTime() - startedAt) / 1_000_000L;
}
bms-back/bms-system/src/main/java/com/bms/price/oracle/TabOrderMonitorService.java
+78 -19
View File
@@ -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)
bms-back/bms-system/src/main/java/com/bms/price/oracle/TabOrderRow.java
+30
View File
@@ -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;
}
}
bms-back/bms-system/src/main/java/com/bms/price/service/IPrcBillingTaskService.java
+7
View File
@@ -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);
bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcBillingTaskServiceImpl.java
+102 -51
View File
@@ -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)
{
bms-back/bms-system/src/main/java/com/bms/price/task/TabOrderSyncAsyncRunner.java
+52 -22
View File
@@ -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;
bms-back/bms-system/src/main/resources/mapper/price/OracleUpstreamOrderMapper.xml
+4 -4
View File
@@ -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=")">
bms-front/src/api/ar/bill.js
+6 -1
View File
@@ -42,7 +42,12 @@ export function retryArBillGenerateTask(taskId) {
}
export function previewArBill(query) {
return request({ url: '/ar/bill/generate/preview', method: 'get', params: query })
return request({
url: '/ar/bill/generate/preview',
method: 'post',
data: query,
timeout: 60000
})
}
export function confirmArBill(arBillId, data = {}) {
bms-front/src/views/monitor/orderSource/components/OrderSourceSyncDialog.vue
+7 -3
View File
@@ -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) {
bms-front/src/views/monitor/orderSource/index.vue
+18 -6
View File
@@ -133,7 +133,7 @@ const searchColumns = [
label: '单号',
prop: 'orderIdentifiers',
component: 'textarea',
placeholder: '订单号/运单号/转单号，一行一个，支持数万个',
placeholder: '订单号/运单号/转单号，一行一个，单次最多 1 万个',
props: {
rows: 2,
autosize: { minRows: 1, maxRows: 4 },
@@ -217,8 +217,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
@@ -267,11 +267,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(() => {})
}