From 557d43a0b850a5db5bf8517b0a018909269a2523 Mon Sep 17 00:00:00 2001 From: wuweihua <88888888> Date: Wed, 4 Nov 2026 19:38:13 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E8=BF=9B=E5=BA=A6=E6=9D=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/price/BillingController.java | 8 + .../main/resources/i18n/messages.properties | 3 + .../domain/BatchAddIncomeBillFeeTask.java | 1 + .../price/service/IPrcOrderBillService.java | 3 + .../service/impl/PrcOrderBillServiceImpl.java | 19 + .../PrcOrderBillLocalRebillAsyncRunner.java | 110 +++++- bms-front/src/api/price/billing.js | 37 +- bms-front/src/views/price/orderBill/index.vue | 352 ++++++++++++++++-- 8 files changed, 483 insertions(+), 50 deletions(-) diff --git a/bms-back/bms-admin/src/main/java/com/bms/web/controller/price/BillingController.java b/bms-back/bms-admin/src/main/java/com/bms/web/controller/price/BillingController.java index 4e980bf6..cc8a5d03 100644 --- a/bms-back/bms-admin/src/main/java/com/bms/web/controller/price/BillingController.java +++ b/bms-back/bms-admin/src/main/java/com/bms/web/controller/price/BillingController.java @@ -305,6 +305,14 @@ public class BillingController extends BaseController return success(task); } + @PreAuthorize("@ss.hasPermi('price:orderBill:rebill')") + @Log(title = "Order bill local rebill cancel", businessType = BusinessType.UPDATE) + @PostMapping("/orderBill/rebill/task/{taskId}/cancel") + public AjaxResult cancelLocalRebill(@PathVariable("taskId") String taskId) + { + return success(prcOrderBillService.cancelLocalRebill(taskId)); + } + /** * Order no options for remote-search dropdown (bill list multi search). */ diff --git a/bms-back/bms-admin/src/main/resources/i18n/messages.properties b/bms-back/bms-admin/src/main/resources/i18n/messages.properties index 62b2bcf4..5f3e8d29 100644 --- a/bms-back/bms-admin/src/main/resources/i18n/messages.properties +++ b/bms-back/bms-admin/src/main/resources/i18n/messages.properties @@ -232,3 +232,6 @@ price.bill.export.notReady=导出文件尚未生成或已失败 price.bill.export.fileMissing=导出文件不存在或路径非法 price.set.customer.batch.empty=请选择要绑定的客户 price.set.customer.batch.done=批量绑定完成：共 {0} 个客户，新绑定 {1} 个，更换 {2} 个，跳过 {3} 个 +price.bill.recalc.taskCancelled=本地重新计费已取消 +price.bill.recalc.taskCancelNotRunning=任务已结束或不存在，无法取消 +price.bill.recalc.taskIdRequired=任务ID不能为空 diff --git a/bms-back/bms-system/src/main/java/com/bms/price/domain/BatchAddIncomeBillFeeTask.java b/bms-back/bms-system/src/main/java/com/bms/price/domain/BatchAddIncomeBillFeeTask.java index efe84a48..322c386f 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/domain/BatchAddIncomeBillFeeTask.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/domain/BatchAddIncomeBillFeeTask.java @@ -11,6 +11,7 @@ public class BatchAddIncomeBillFeeTask implements Serializable public static final String STATUS_RUNNING = "RUNNING"; public static final String STATUS_SUCCESS = "SUCCESS"; public static final String STATUS_FAIL = "FAIL"; + public static final String STATUS_CANCELLED = "CANCELLED"; private String taskId; private String status; 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 a0747cf0..b5db8a2a 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 @@ -166,6 +166,9 @@ public interface IPrcOrderBillService /** Poll Redis progress for local rebill task. */ BatchAddIncomeBillFeeTask getLocalRebillTask(String taskId); + /** Cooperative cancel for a running local rebill task. */ + BatchAddIncomeBillFeeTask cancelLocalRebill(String taskId); + /** Data-scoped bill_id list for local rebill (same filters as order-bill list). */ List selectBillIdsForLocalRebill(PrcOrderBill query); 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 2b65f2ab..5b38b12f 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 @@ -972,6 +972,25 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService return localRebillAsyncRunner.get(taskId); } + @Override + public BatchAddIncomeBillFeeTask cancelLocalRebill(String taskId) + { + if (StringUtils.isEmpty(taskId)) + { + throw new ServiceException(MessageUtils.message("price.bill.recalc.taskIdRequired")); + } + if (localRebillAsyncRunner == null) + { + throw new ServiceException(MessageUtils.message("price.bill.recalc.taskNotExists")); + } + BatchAddIncomeBillFeeTask task = localRebillAsyncRunner.requestCancel(taskId.trim()); + if (task == null) + { + throw new ServiceException(MessageUtils.message("price.bill.recalc.taskNotExists")); + } + return task; + } + @Override public String submitOrderBillExport(String exportType, PrcOrderBill query) { 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 98591ef9..eabb1923 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 @@ -6,12 +6,14 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,8 +34,9 @@ import com.bms.price.search.TabOrderSyncEsLog; import com.bms.price.service.IPrcOrderBillService; /** - * 仅根据 prc_order_bill 异步本地重计费（不走 Oracle / 不走 MQ）。 - * 大批量按 {@link #PAGE_SIZE} 分页处理；每一页在短生命周期线程池中执行。 + * 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. */ @Component public class PrcOrderBillLocalRebillAsyncRunner @@ -48,6 +51,8 @@ public class PrcOrderBillLocalRebillAsyncRunner public static final int MAX_THREADS = 16; private static final int FAIL_DETAIL_MAX = 200; + private final ConcurrentHashMap cancelFlags = new ConcurrentHashMap(); + @Autowired @Lazy private IPrcOrderBillService orderBillService; @@ -79,6 +84,7 @@ public class PrcOrderBillLocalRebillAsyncRunner StringBuilder failTips = new StringBuilder(); List> failDetails = Collections.synchronizedList(new ArrayList>()); int threads = clampThreads(billingProperties == null ? null : billingProperties.getOrderSourceSyncThreads()); + cancelFlags.computeIfAbsent(taskId, k -> new AtomicBoolean(false)); ExecutorService pool = newWorkerPool(threads, "local-rebill"); try { @@ -89,9 +95,20 @@ public class PrcOrderBillLocalRebillAsyncRunner saveDoneLog(taskId, request, total, 0, 0, threads, failDetails, "FAIL", "no bills"); return; } + if (isCancelRequested(taskId)) + { + finishCancelled(taskId, request, total, success.get(), fail.get(), threads, failDetails); + return; + } int processed = 0; + boolean cancelled = false; for (int from = 0; from < total; from += PAGE_SIZE) { + 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()); @@ -99,6 +116,10 @@ public class PrcOrderBillLocalRebillAsyncRunner { final String one = billId; jobs.add(() -> { + if (isCancelRequested(taskId)) + { + return Boolean.FALSE; + } try { orderBillService.rebillLocalByBillId(one); @@ -118,11 +139,21 @@ public class PrcOrderBillLocalRebillAsyncRunner } invokeAllQuiet(pool, jobs); processed = to; + if (isCancelRequested(taskId)) + { + cancelled = true; + break; + } int percent = Math.min(99, (int) Math.floor(100.0 * processed / total)); put(taskId, BatchAddIncomeBillFeeTask.STATUS_RUNNING, percent, "processing " + processed + "/" + total + " page=" + ((from / PAGE_SIZE) + 1), null, total, success.get(), fail.get()); } + if (cancelled || isCancelRequested(taskId)) + { + finishCancelled(taskId, request, total, success.get(), fail.get(), threads, failDetails); + return; + } String msg = "success=" + success.get() + ", fail=" + fail.get(); String err = fail.get() > 0 && failTips.length() > 0 ? failTips.toString() : null; put(taskId, BatchAddIncomeBillFeeTask.STATUS_SUCCESS, 100, msg, err, @@ -135,6 +166,11 @@ public class PrcOrderBillLocalRebillAsyncRunner } catch (Exception e) { + if (isCancelRequested(taskId)) + { + finishCancelled(taskId, request, total, success.get(), fail.get(), threads, failDetails); + return; + } log.error("local rebill aborted taskId={}", taskId, e); String msg = errorMessage(e); put(taskId, BatchAddIncomeBillFeeTask.STATUS_FAIL, 100, "failed", msg, @@ -144,10 +180,62 @@ public class PrcOrderBillLocalRebillAsyncRunner } finally { + cancelFlags.remove(taskId); shutdownPool(pool); } } + /** + * Request cooperative cancel. Returns current task snapshot (may still be RUNNING until page ends). + */ + public BatchAddIncomeBillFeeTask requestCancel(String taskId) + { + if (StringUtils.isEmpty(taskId)) + { + return null; + } + BatchAddIncomeBillFeeTask task = get(taskId); + if (task == null) + { + return null; + } + String st = task.getStatus(); + if (BatchAddIncomeBillFeeTask.STATUS_SUCCESS.equals(st) + || BatchAddIncomeBillFeeTask.STATUS_FAIL.equals(st) + || BatchAddIncomeBillFeeTask.STATUS_CANCELLED.equals(st)) + { + return task; + } + cancelFlags.computeIfAbsent(taskId, k -> new AtomicBoolean(false)).set(true); + // Keep RUNNING so clients keep polling until finishCancelled writes CANCELLED. + put(taskId, BatchAddIncomeBillFeeTask.STATUS_RUNNING, + Math.max(task.getPercent(), 1), + "cancelling", + null, + task.getTotal(), task.getSuccess(), task.getFail()); + log.info("local rebill cancel requested taskId={} success={} fail={} total={}", + taskId, task.getSuccess(), task.getFail(), task.getTotal()); + return get(taskId); + } + + private void finishCancelled(String taskId, LocalRebillRequest request, int total, int ok, int fail, + int threads, List> failDetails) + { + int percent = total <= 0 ? 100 : Math.min(100, (int) Math.floor(100.0 * (ok + fail) / total)); + String msg = "cancelled success=" + ok + ", fail=" + fail; + put(taskId, BatchAddIncomeBillFeeTask.STATUS_CANCELLED, Math.max(percent, 1), msg, + "cancelled by user", total, ok, fail); + saveDoneLog(taskId, request, total, ok, fail, threads, failDetails, + EsBizLogSupport.STATUS_FAIL, "cancelled by user"); + log.info("local rebill cancelled taskId={} total={} success={} fail={}", taskId, total, ok, fail); + } + + private boolean isCancelRequested(String taskId) + { + AtomicBoolean flag = cancelFlags.get(taskId); + return flag != null && flag.get(); + } + private void saveDoneLog(String taskId, LocalRebillRequest request, int total, int ok, int fail, int threads, List> failDetails, String status, String errorMessage) { @@ -209,6 +297,24 @@ public class PrcOrderBillLocalRebillAsyncRunner public void put(String taskId, String status, int percent, String message, String error, int total, int success, int fail) { + BatchAddIncomeBillFeeTask existing = get(taskId); + if (existing != null + && BatchAddIncomeBillFeeTask.STATUS_CANCELLED.equals(existing.getStatus()) + && !BatchAddIncomeBillFeeTask.STATUS_CANCELLED.equals(status)) + { + // Keep cancelled: do not let a late RUNNING overwrite after user cancel. + BatchAddIncomeBillFeeTask keep = new BatchAddIncomeBillFeeTask(); + keep.setTaskId(taskId); + keep.setStatus(BatchAddIncomeBillFeeTask.STATUS_CANCELLED); + keep.setPercent(Math.max(existing.getPercent(), percent)); + keep.setMessage(existing.getMessage()); + keep.setError(existing.getError()); + keep.setTotal(total > 0 ? total : existing.getTotal()); + keep.setSuccess(Math.max(existing.getSuccess(), success)); + keep.setFail(Math.max(existing.getFail(), fail)); + redisCache.setCacheObject(CACHE_KEY_PREFIX + taskId, keep, TTL_MINUTES, TimeUnit.MINUTES); + return; + } BatchAddIncomeBillFeeTask task = new BatchAddIncomeBillFeeTask(); task.setTaskId(taskId); task.setStatus(status); diff --git a/bms-front/src/api/price/billing.js b/bms-front/src/api/price/billing.js index 17e42223..e4f6ea49 100644 --- a/bms-front/src/api/price/billing.js +++ b/bms-front/src/api/price/billing.js @@ -102,23 +102,29 @@ export function getLocalRebillTask(taskId) { }) } +export function cancelLocalRebill(taskId) { + return request({ + url: '/price/billing/orderBill/rebill/task/' + taskId + '/cancel', + method: 'post' + }) +} + /** - * Async local rebill; poll until done. - * @param {object} data LocalRebillRequest + * Poll local rebill task until terminal status. + * @param {string} taskId * @param {(task: object) => void} [onProgress] */ -export async function localRebillAsync(data, onProgress) { - const submitRes = await submitLocalRebill(data) - const d = submitRes && submitRes.data - const taskId = d && (d.taskId || d) - if (!taskId) throw new Error('missing local rebill taskId') +export async function pollLocalRebillTask(taskId, onProgress) { + const id = String(taskId || '') + if (!id) throw new Error('missing local rebill taskId') const started = Date.now() const maxMs = 30 * 60 * 1000 while (Date.now() - started < maxMs) { - const res = await getLocalRebillTask(String(taskId)) + const res = await getLocalRebillTask(id) const task = res.data || {} if (typeof onProgress === 'function') onProgress(task) if (task.status === 'SUCCESS') return { task } + if (task.status === 'CANCELLED') return { task, cancelled: true } if (task.status === 'FAIL') { const err = new Error(task.error || task.message || 'local rebill failed') err.task = task @@ -129,6 +135,21 @@ export async function localRebillAsync(data, onProgress) { throw new Error('local rebill timeout') } +/** + * Async local rebill; poll until done. + * @param {object} data LocalRebillRequest + * @param {(task: object) => void} [onProgress] + * @param {{ getTaskId?: (taskId: string) => void }} [options] + */ +export async function localRebillAsync(data, onProgress, options) { + const submitRes = await submitLocalRebill(data) + const d = submitRes && submitRes.data + const taskId = d && (d.taskId || d) + if (!taskId) throw new Error('missing local rebill taskId') + if (options && typeof options.getTaskId === 'function') options.getTaskId(String(taskId)) + return pollLocalRebillTask(String(taskId), onProgress) +} + /** Receivable bill detail */ export function getBill(billId) { return request({ diff --git a/bms-front/src/views/price/orderBill/index.vue b/bms-front/src/views/price/orderBill/index.vue index 5087d30a..04ca336a 100644 --- a/bms-front/src/views/price/orderBill/index.vue +++ b/bms-front/src/views/price/orderBill/index.vue @@ -1,5 +1,19 @@