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 01aa1b3c..ab59f058 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 @@ -1851,6 +1851,36 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService } dedup.put(orderNo, row); } + if (dedup.size() > CHARGE_WEIGHT_IMPORT_MAX) + { + throw new ServiceException("导入行数超限，最多 " + + CHARGE_WEIGHT_IMPORT_MAX + " 条"); + } + + // Batch load bills with data-scope (chunked IN), avoid N+1 select + getBill + java.util.Map billByOrderNo = new java.util.HashMap(); + List allOrderNos = new ArrayList(dedup.keySet()); + final int chunkSize = 500; + for (int from = 0; from < allOrderNos.size(); from += chunkSize) + { + int to = Math.min(from + chunkSize, allOrderNos.size()); + List chunk = new ArrayList(allOrderNos.subList(from, to)); + PrcOrderBill query = new PrcOrderBill(); + query.setOrderNos(chunk); + List scoped = self.selectBillList(query); + if (scoped != null) + { + for (PrcOrderBill bill : scoped) + { + if (bill == null || StringUtils.isEmpty(bill.getOrderNo())) + { + continue; + } + billByOrderNo.put(bill.getOrderNo().trim(), bill); + } + } + } + List items = new ArrayList(); for (java.util.Map.Entry entry : dedup.entrySet()) { @@ -1867,13 +1897,11 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService { throw new ServiceException("客户不能为空"); } - PrcOrderBill bill = orderBillMapper.selectPrcOrderBillByOrderNo(orderNo); + PrcOrderBill bill = billByOrderNo.get(orderNo); if (bill == null) { throw new ServiceException("订单不存在"); } - // data-scope in request thread - self.getBill(bill.getBillId()); if (!customerMatches(bill, customer)) { throw new ServiceException("客户与订单不匹配"); @@ -1903,11 +1931,6 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService { throw new ServiceException("导入数据不能为空"); } - if (items.size() > CHARGE_WEIGHT_IMPORT_MAX) - { - throw new ServiceException("导入行数超限，最多 " - + CHARGE_WEIGHT_IMPORT_MAX + " 条"); - } String taskId = IdUtils.fastSimpleUUID(); int total = items.size() + initialFail; chargeWeightImportAsyncRunner.put(taskId, ChargeWeightImportTask.STATUS_PENDING, 0, "queued", diff --git a/bms-back/bms-system/src/main/java/com/bms/price/task/PrcChargeWeightImportAsyncRunner.java b/bms-back/bms-system/src/main/java/com/bms/price/task/PrcChargeWeightImportAsyncRunner.java index f197b4d4..568c2f14 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/task/PrcChargeWeightImportAsyncRunner.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/task/PrcChargeWeightImportAsyncRunner.java @@ -1,8 +1,13 @@ package com.bms.price.task; import java.math.BigDecimal; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -15,7 +20,7 @@ import com.bms.price.domain.ChargeWeightImportTask; import com.bms.price.service.IPrcOrderBillService; /** - * Async runner: import charge_weight updates and rebill each order. + * Async runner: import charge_weight updates and rebill each order (parallel). */ @Component public class PrcChargeWeightImportAsyncRunner @@ -24,6 +29,10 @@ public class PrcChargeWeightImportAsyncRunner public static final String CACHE_KEY_PREFIX = "price:billing:chargeWeight:import:task:"; public static final int TTL_MINUTES = 60; + /** Parallel rebill workers per import task. */ + private static final int REBILL_CONCURRENCY = 8; + /** Redis progress write every N finished rows. */ + private static final int PROGRESS_EVERY = 10; @Autowired @Lazy @@ -32,28 +41,25 @@ public class PrcChargeWeightImportAsyncRunner @Autowired private RedisCache redisCache; - /** - * @param items pre-validated rows with billId filled in remark field unused; - * use orderNo + chargeWeight; billId stored in customer field as billId carrier - * -- better use a dedicated work item. See run(List of Object[]). - */ @Async("threadPoolTaskExecutor") public void run(String taskId, List items, int initialFail) { - int total = (items == null ? 0 : items.size()) + Math.max(initialFail, 0); - int success = 0; - int fail = Math.max(initialFail, 0); + int workSize = items == null ? 0 : items.size(); + int total = workSize + Math.max(initialFail, 0); + AtomicInteger success = new AtomicInteger(0); + AtomicInteger fail = new AtomicInteger(Math.max(initialFail, 0)); + AtomicInteger doneWork = new AtomicInteger(0); StringBuilder failTips = new StringBuilder(); + Object tipsLock = new Object(); try { - put(taskId, ChargeWeightImportTask.STATUS_RUNNING, 1, "running", null, total, 0, fail); - int workSize = items == null ? 0 : items.size(); + put(taskId, ChargeWeightImportTask.STATUS_RUNNING, 1, "running", null, total, 0, fail.get()); if (workSize == 0) { - if (fail > 0) + if (fail.get() > 0) { put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, - "success=0, fail=" + fail, "all rows failed validation", total, 0, fail); + "success=0, fail=" + fail.get(), "all rows failed validation", total, 0, fail.get()); } else { @@ -61,55 +67,102 @@ public class PrcChargeWeightImportAsyncRunner } return; } - for (int i = 0; i < workSize; i++) + + int poolSize = Math.min(REBILL_CONCURRENCY, workSize); + ExecutorService pool = Executors.newFixedThreadPool(poolSize); + List> futures = new ArrayList>(workSize); + try { - ChargeWeightImportWorkItem item = items.get(i); + for (int i = 0; i < workSize; i++) + { + final ChargeWeightImportWorkItem item = items.get(i); + futures.add(pool.submit(new Runnable() + { + @Override + public void run() + { + try + { + orderBillService.updateChargeWeightByBillId(item.getBillId(), item.getChargeWeight()); + success.incrementAndGet(); + } + catch (Exception e) + { + fail.incrementAndGet(); + synchronized (tipsLock) + { + if (failTips.length() < 800) + { + String tip = e.getMessage(); + if (StringUtils.isEmpty(tip)) + { + tip = e.getClass().getSimpleName(); + } + if (failTips.length() > 0) + { + failTips.append("; "); + } + failTips.append(item.getOrderNo()).append(":").append(tip); + } + } + log.warn("charge weight import failed taskId={} orderNo={}", + taskId, item.getOrderNo(), e); + } + finally + { + int finished = doneWork.incrementAndGet(); + if (finished % PROGRESS_EVERY == 0 || finished == workSize) + { + int done = finished + Math.max(initialFail, 0); + int percent = Math.min(99, + (int) Math.floor(100.0 * done / Math.max(total, 1))); + put(taskId, ChargeWeightImportTask.STATUS_RUNNING, percent, + "processing " + done + "/" + total, null, total, + success.get(), fail.get()); + } + } + } + })); + } + for (Future f : futures) + { + f.get(); + } + } + finally + { + pool.shutdown(); try { - orderBillService.updateChargeWeightByBillId(item.getBillId(), item.getChargeWeight()); - success++; + pool.awaitTermination(30, TimeUnit.MINUTES); } - catch (Exception e) + catch (InterruptedException ie) { - fail++; - if (failTips.length() < 800) - { - String tip = e.getMessage(); - if (StringUtils.isEmpty(tip)) - { - tip = e.getClass().getSimpleName(); - } - if (failTips.length() > 0) - { - failTips.append("; "); - } - failTips.append(item.getOrderNo()).append(":").append(tip); - } - log.warn("charge weight import failed taskId={} orderNo={}", taskId, item.getOrderNo(), e); + Thread.currentThread().interrupt(); + pool.shutdownNow(); } - int done = i + 1 + Math.max(initialFail, 0); - int percent = Math.min(99, (int) Math.floor(100.0 * done / Math.max(total, 1))); - put(taskId, ChargeWeightImportTask.STATUS_RUNNING, percent, - "processing " + done + "/" + total, null, total, success, fail); } - String msg = "success=" + success + ", fail=" + fail; - if (fail > 0 && failTips.length() > 0) + + String msg = "success=" + success.get() + ", fail=" + fail.get(); + if (fail.get() > 0 && failTips.length() > 0) { put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, msg, failTips.toString(), - total, success, fail); + total, success.get(), fail.get()); } else { - put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, msg, null, total, success, fail); + put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, msg, null, + total, success.get(), fail.get()); } - log.info("charge weight import done taskId={} total={} success={} fail={}", - taskId, total, success, fail); + log.info("charge weight import done taskId={} total={} success={} fail={} concurrency={}", + taskId, Integer.valueOf(total), Integer.valueOf(success.get()), + Integer.valueOf(fail.get()), Integer.valueOf(poolSize)); } catch (Exception e) { log.error("charge weight import aborted taskId={}", taskId, e); put(taskId, ChargeWeightImportTask.STATUS_FAIL, 100, "failed", errorMessage(e), - total, success, fail); + total, success.get(), fail.get()); } }