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

优化计费速度

Browse Source
This commit is contained in:
wuweihua
2026-11-05 16:58:30 +08:00
parent 36433863af
commit 44df6b1a1b
7 changed files with 52 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bms-back/bms-admin/src/main/resources/application-local.yml
+2 -2
View File
@@ -38,8 +38,8 @@ spring:
initialSize: 5
# 最小连接池数量
minIdle: 10
# 最大连接池数量
maxActive: 20
# 最大连接池数量（local rebill 16 workers + UI/MQ headroom）
maxActive: 40
# 配置获取连接等待超时的时间
maxWait: 60000
# 配置连接超时时间
bms-back/bms-admin/src/main/resources/application-prod.yml
+2 -2
View File
@@ -51,8 +51,8 @@ spring:
initialSize: 5
# 最小连接池数量
minIdle: 10
# 最大连接池数量
maxActive: 20
# 最大连接池数量（local rebill 16 workers + UI/MQ headroom）
maxActive: 40
# 配置获取连接等待超时的时间
maxWait: 60000
# 配置连接超时时间
bms-back/bms-admin/src/main/resources/application.yml
+2
View File
@@ -40,6 +40,8 @@ bms:
# 同步已直接复用订单行（跳过每单 Oracle 公网查询），瓶颈在 MySQL 计费写入，
# 同客户行锁冲突由 billWithDbRetry 3 次重试兜底，16 路实测吞吐最高。
order-source-sync-threads: 16
# Order-list local rebill workers (2~24). Pair with Druid maxActive >= threads + 8.
local-rebill-threads: 16
# Night auto-rebill: hours 0-8, pull 500 (all statuses) -> bill -> next 500 until empty / 09:00.
# Empty page => no billing.
night-retry:
bms-back/bms-system/src/main/java/com/bms/price/config/BillingProperties.java
+15
View File
@@ -20,6 +20,11 @@ public class BillingProperties
private int consumerThreads = 8;
/** Concurrent threads for monitor order-source sync */
private int orderSourceSyncThreads = 8;
/**
* Concurrent threads for order-list local rebill (2~24).
* Keep below Druid maxActive with headroom for other traffic.
*/
private int localRebillThreads = 16;
/** Night window auto-retry of prc_billing_task (all statuses) */
private NightRetry nightRetry = new NightRetry();
private Oracle oracle = new Oracle();
@@ -84,6 +89,16 @@ public class BillingProperties
this.orderSourceSyncThreads = orderSourceSyncThreads;
}
public int getLocalRebillThreads()
{
return localRebillThreads;
}
public void setLocalRebillThreads(int localRebillThreads)
{
this.localRebillThreads = localRebillThreads;
}
public NightRetry getNightRetry()
{
return nightRetry;
bms-back/bms-system/src/main/java/com/bms/price/service/IPrcOrderBillService.java
+1 -1
View File
@@ -159,7 +159,7 @@ public interface IPrcOrderBillService
/**
* Async local rebill by selected billIds or list filters. Returns taskId for progress poll.
* Large sets: only bill_id loaded, then processed in pages of 500 with parallel workers.
* Large sets: only bill_id loaded, then processed by continuous parallel workers.
*/
String submitLocalRebill(LocalRebillRequest request);
bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java
+2 -2
View File
@@ -183,8 +183,8 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService
private static final int BATCH_ADD_FEE_PREVIEW_LIMIT = 20;
private static final int CHARGE_WEIGHT_IMPORT_MAX = 5000;
private static final int RECALC_PRICE_MAX = 5000;
/** Soft max for one local-rebill job (id-only load + paged parallel). */
private static final int LOCAL_REBILL_MAX = 100000;
/** Soft max for one local-rebill job (id-only load + continuous workers). */
private static final int LOCAL_REBILL_MAX = 200000;
/** Soft max for one order/billing list Excel export. */
private static final int ORDER_BILL_EXPORT_MAX = 50000;
bms-back/bms-system/src/main/java/com/bms/price/task/PrcOrderBillLocalRebillAsyncRunner.java
+28 -42
View File
@@ -36,8 +36,7 @@ import com.bms.price.service.IPrcOrderBillService;
/**
* Async local rebill from prc_order_bill only (no Oracle / MQ).
* Large jobs page by {@link #PAGE_SIZE}; each page runs in a short-lived pool.
* Cancel is cooperative: current in-flight bills finish, remaining bills are skipped.
* Continuous worker pull (no page barrier); cancel skips remaining billIds.
*/
@Component
public class PrcOrderBillLocalRebillAsyncRunner
@@ -46,13 +45,11 @@ public class PrcOrderBillLocalRebillAsyncRunner
public static final String CACHE_KEY_PREFIX = "price:billing:localRebill:task:";
public static final int TTL_MINUTES = 120;
/** Bill this many, wait for page done, then next page. */
public static final int PAGE_SIZE = 500;
public static final int MIN_THREADS = 2;
public static final int MAX_THREADS = 16;
public static final int MAX_THREADS = 24;
private static final int FAIL_DETAIL_MAX = 200;
/** Publish Redis progress at most this often while a page is running. */
private static final int PROGRESS_EVERY_N = 20;
/** Publish Redis progress at most this often while workers run. */
private static final int PROGRESS_EVERY_N = 50;
private static final long PROGRESS_MIN_INTERVAL_MS = 500L;
private final ConcurrentHashMap<String, AtomicBoolean> cancelFlags = new ConcurrentHashMap<String, AtomicBoolean>();
@@ -86,10 +83,12 @@ public class PrcOrderBillLocalRebillAsyncRunner
AtomicInteger success = new AtomicInteger(0);
AtomicInteger fail = new AtomicInteger(0);
AtomicInteger done = new AtomicInteger(0);
AtomicInteger nextIndex = new AtomicInteger(0);
AtomicLong lastProgressAt = new AtomicLong(0L);
StringBuilder failTips = new StringBuilder();
List<Map<String, Object>> failDetails = Collections.synchronizedList(new ArrayList<Map<String, Object>>());
int threads = clampThreads(billingProperties == null ? null : billingProperties.getOrderSourceSyncThreads());
int configured = billingProperties == null ? 16 : billingProperties.getLocalRebillThreads();
int threads = clampThreads(Integer.valueOf(configured));
cancelFlags.computeIfAbsent(taskId, k -> new AtomicBoolean(false));
ExecutorService pool = newWorkerPool(threads, "local-rebill");
try
@@ -107,30 +106,24 @@ public class PrcOrderBillLocalRebillAsyncRunner
finishCancelled(taskId, request, total, success.get(), fail.get(), threads, failDetails);
return;
}
boolean cancelled = false;
for (int from = 0; from < total; from += PAGE_SIZE)
// Continuous pull: N workers grab next billId — no page barrier idle wait.
final List<String> ids = billIds;
List<Callable<Boolean>> workers = new ArrayList<Callable<Boolean>>(threads);
for (int w = 0; w < threads; w++)
{
if (isCancelRequested(taskId))
{
cancelled = true;
break;
}
int to = Math.min(from + PAGE_SIZE, total);
List<String> page = billIds.subList(from, to);
List<Callable<Boolean>> jobs = new ArrayList<Callable<Boolean>>(page.size());
for (String billId : page)
{
final String one = billId;
jobs.add(() -> {
if (isCancelRequested(taskId))
workers.add(() -> {
while (!isCancelRequested(taskId))
{
int i = nextIndex.getAndIncrement();
if (i >= total)
{
return Boolean.FALSE;
return Boolean.TRUE;
}
String one = ids.get(i);
try
{
orderBillService.rebillLocalByBillId(one);
success.incrementAndGet();
return Boolean.TRUE;
}
catch (Exception e)
{
@@ -139,26 +132,20 @@ public class PrcOrderBillLocalRebillAsyncRunner
collectFailDetail(failDetails, one, e);
log.warn("local rebill failed taskId={} billId={} msg={}",
taskId, one, e.getMessage());
return Boolean.FALSE;
}
finally
{
int finished = done.incrementAndGet();
maybePublishProgress(taskId, total, finished, success.get(), fail.get(), lastProgressAt);
maybePublishProgress(taskId, total, finished, success.get(), fail.get(),
lastProgressAt);
}
});
}
invokeAllQuiet(pool, jobs);
if (isCancelRequested(taskId))
{
cancelled = true;
break;
}
// Page boundary: always flush latest counts (even if throttle skipped mid-page).
publishProgress(taskId, total, done.get(), success.get(), fail.get());
lastProgressAt.set(System.currentTimeMillis());
}
return Boolean.FALSE;
});
}
if (cancelled || isCancelRequested(taskId))
invokeAllQuiet(pool, workers);
publishProgress(taskId, total, done.get(), success.get(), fail.get());
if (isCancelRequested(taskId))
{
finishCancelled(taskId, request, total, success.get(), fail.get(), threads, failDetails);
return;
@@ -170,8 +157,8 @@ public class PrcOrderBillLocalRebillAsyncRunner
String doneStatus = fail.get() > 0 ? EsBizLogSupport.STATUS_FAIL : EsBizLogSupport.STATUS_SUCCESS;
String doneErr = fail.get() > 0 ? ("fail=" + fail.get() + ", ok=" + success.get()) : null;
saveDoneLog(taskId, request, total, success.get(), fail.get(), threads, failDetails, doneStatus, doneErr);
log.info("local rebill done taskId={} total={} success={} fail={} threads={} pageSize={}",
taskId, total, success.get(), fail.get(), threads, PAGE_SIZE);
log.info("local rebill done taskId={} total={} success={} fail={} threads={}",
taskId, total, success.get(), fail.get(), threads);
}
catch (Exception e)
{
@@ -255,7 +242,6 @@ public class PrcOrderBillLocalRebillAsyncRunner
resp.put("fail", Integer.valueOf(fail));
resp.put("success", Integer.valueOf(ok));
resp.put("threads", Integer.valueOf(threads));
resp.put("pageSize", Integer.valueOf(PAGE_SIZE));
if (failDetails != null && !failDetails.isEmpty())
{
resp.put("failDetails", new ArrayList<Map<String, Object>>(failDetails));