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-04 19:38:13 +08:00
parent 942d5932ae
commit 557d43a0b8
8 changed files with 483 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bms-back/bms-admin/src/main/java/com/bms/web/controller/price/BillingController.java
+8
View File
@@ -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).
*/
bms-back/bms-admin/src/main/resources/i18n/messages.properties
+3
View File
@@ -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不能为空
bms-back/bms-system/src/main/java/com/bms/price/domain/BatchAddIncomeBillFeeTask.java
+1
View File
@@ -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;
bms-back/bms-system/src/main/java/com/bms/price/service/IPrcOrderBillService.java
+3
View File
@@ -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<String> selectBillIdsForLocalRebill(PrcOrderBill query);
bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java
+19
View File
@@ -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)
{
bms-back/bms-system/src/main/java/com/bms/price/task/PrcOrderBillLocalRebillAsyncRunner.java
+108 -2
View File
@@ -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<String, AtomicBoolean> cancelFlags = new ConcurrentHashMap<String, AtomicBoolean>();
@Autowired
@Lazy
private IPrcOrderBillService orderBillService;
@@ -79,6 +84,7 @@ public class PrcOrderBillLocalRebillAsyncRunner
StringBuilder failTips = new StringBuilder();
List<Map<String, Object>> failDetails = Collections.synchronizedList(new ArrayList<Map<String, Object>>());
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<String> page = billIds.subList(from, to);
List<Callable<Boolean>> jobs = new ArrayList<Callable<Boolean>>(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<Map<String, Object>> 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<Map<String, Object>> 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);