diff --git a/bms-back/bms-admin/src/main/resources/application-local.yml b/bms-back/bms-admin/src/main/resources/application-local.yml index d9e07474..f9a2f83b 100644 --- a/bms-back/bms-admin/src/main/resources/application-local.yml +++ b/bms-back/bms-admin/src/main/resources/application-local.yml @@ -38,8 +38,8 @@ spring: initialSize: 5 # 最小连接池数量 minIdle: 10 - # 最大连接池数量 - maxActive: 20 + # 最大连接池数量（local rebill 16 workers + UI/MQ headroom） + maxActive: 40 # 配置获取连接等待超时的时间 maxWait: 60000 # 配置连接超时时间 diff --git a/bms-back/bms-admin/src/main/resources/application-prod.yml b/bms-back/bms-admin/src/main/resources/application-prod.yml index 1ea85a50..ccf6cb81 100644 --- a/bms-back/bms-admin/src/main/resources/application-prod.yml +++ b/bms-back/bms-admin/src/main/resources/application-prod.yml @@ -51,8 +51,8 @@ spring: initialSize: 5 # 最小连接池数量 minIdle: 10 - # 最大连接池数量 - maxActive: 20 + # 最大连接池数量（local rebill 16 workers + UI/MQ headroom） + maxActive: 40 # 配置获取连接等待超时的时间 maxWait: 60000 # 配置连接超时时间 diff --git a/bms-back/bms-admin/src/main/resources/application.yml b/bms-back/bms-admin/src/main/resources/application.yml index a61fcc06..29427a10 100644 --- a/bms-back/bms-admin/src/main/resources/application.yml +++ b/bms-back/bms-admin/src/main/resources/application.yml @@ -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: diff --git a/bms-back/bms-system/src/main/java/com/bms/price/config/BillingProperties.java b/bms-back/bms-system/src/main/java/com/bms/price/config/BillingProperties.java index fd8ecb2e..7dd9c9c0 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/config/BillingProperties.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/config/BillingProperties.java @@ -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; diff --git a/bms-back/bms-system/src/main/java/com/bms/price/service/IPrcOrderBillService.java b/bms-back/bms-system/src/main/java/com/bms/price/service/IPrcOrderBillService.java index b5db8a2a..f8139c2e 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/service/IPrcOrderBillService.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/service/IPrcOrderBillService.java @@ -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); diff --git a/bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java b/bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java index 5b38b12f..f1617014 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java @@ -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; diff --git a/bms-back/bms-system/src/main/java/com/bms/price/task/PrcOrderBillLocalRebillAsyncRunner.java b/bms-back/bms-system/src/main/java/com/bms/price/task/PrcOrderBillLocalRebillAsyncRunner.java index 4ca52106..c2905a14 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/task/PrcOrderBillLocalRebillAsyncRunner.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/task/PrcOrderBillLocalRebillAsyncRunner.java @@ -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 cancelFlags = new ConcurrentHashMap(); @@ -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> failDetails = Collections.synchronizedList(new ArrayList>()); - 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 ids = billIds; + List> workers = new ArrayList>(threads); + for (int w = 0; w < threads; w++) { - if (isCancelRequested(taskId)) - { - cancelled = true; - break; - } - int to = Math.min(from + PAGE_SIZE, total); - List page = billIds.subList(from, to); - List> jobs = new ArrayList>(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>(failDetails));