diff --git a/bms-back/bms-admin/src/main/java/com/bms/web/controller/monitor/TabOrderMonitorController.java b/bms-back/bms-admin/src/main/java/com/bms/web/controller/monitor/TabOrderMonitorController.java index af1c1752..877e7ca7 100644 --- a/bms-back/bms-admin/src/main/java/com/bms/web/controller/monitor/TabOrderMonitorController.java +++ b/bms-back/bms-admin/src/main/java/com/bms/web/controller/monitor/TabOrderMonitorController.java @@ -1,6 +1,5 @@ package com.bms.web.controller.monitor; -import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -24,17 +23,14 @@ import com.bms.common.core.page.TableDataInfo; import com.bms.common.core.page.TableSupport; import com.bms.common.enums.BusinessType; import com.bms.common.utils.StringUtils; -import com.bms.price.domain.PrcBillingTask; import com.bms.price.domain.PrcBillingTaskRetryTask; import com.bms.price.domain.vo.TabOrderMonitorQuery; import com.bms.price.domain.vo.TabOrderMonitorVo; -import com.bms.price.enums.BillTaskSuccessStatus; import com.bms.price.oracle.TabOrderMonitorService; import com.bms.price.search.BillSearchPage; import com.bms.price.search.EsLogType; import com.bms.price.search.IBillSearchStore; import com.bms.price.search.TabOrderSyncEsLog; -import com.bms.price.service.IPrcBillingTaskService; import com.bms.price.task.TabOrderSyncAsyncRunner; /** @@ -47,9 +43,6 @@ public class TabOrderMonitorController extends BaseController @Autowired private TabOrderMonitorService tabOrderMonitorService; - @Autowired - private IPrcBillingTaskService billingTaskService; - @Autowired private TabOrderSyncAsyncRunner tabOrderSyncAsyncRunner; @@ -126,40 +119,9 @@ public class TabOrderMonitorController extends BaseController { return error("selected rows exceed 200"); } - int ok = 0; - int fail = 0; - List> details = new ArrayList>(); - for (TabOrderMonitorVo row : rows) - { - Map item = new HashMap(); - item.put("orderId", row == null ? null : row.getOrderId()); - item.put("transferOrderId", row == null ? null : row.getTransferOrderId()); - try - { - PrcBillingTask task = billingTaskService.submitAndProcess(TabOrderSyncAsyncRunner.toSubmit(row)); - item.put("successStatus", task == null ? null : task.getSuccessStatus()); - item.put("errorMessage", task == null ? null : task.getErrorMessage()); - if (task != null && BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus())) - { - ok++; - } - else - { - fail++; - } - } - catch (RuntimeException e) - { - fail++; - item.put("successStatus", "ERROR"); - item.put("errorMessage", e.getMessage()); - } - details.add(item); - } - Map data = new HashMap(); - data.put("ok", Integer.valueOf(ok)); - data.put("fail", Integer.valueOf(fail)); - data.put("details", details); + Map data = tabOrderSyncAsyncRunner.syncRowsParallel(rows); + int ok = data.get("ok") instanceof Number ? ((Number) data.get("ok")).intValue() : 0; + int fail = data.get("fail") instanceof Number ? ((Number) data.get("fail")).intValue() : 0; String action = rows.size() == 1 ? "ONE" : "SELECTED"; String bizId = firstOrderBizId(rows, action); String status = fail > 0 && ok == 0 ? "FAIL" : "SUCCESS"; diff --git a/bms-back/bms-admin/src/main/resources/application.yml b/bms-back/bms-admin/src/main/resources/application.yml index ab4abf2b..bfd3c671 100644 --- a/bms-back/bms-admin/src/main/resources/application.yml +++ b/bms-back/bms-admin/src/main/resources/application.yml @@ -26,6 +26,8 @@ bms: # Max messages popped per tick; processed by consumer-threads pool consumer-batch-size: 32 consumer-threads: 8 + # Concurrent threads for monitor order-source sync (2~16) + order-source-sync-threads: 8 oracle: # SLAVE datasource Oracle main-order table (schema.table allowed) order-table: WEISHI_USER.TAB_ORDER 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 7436d6c2..6e56d5b6 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 @@ -18,6 +18,8 @@ public class BillingProperties private int consumerBatchSize = 32; /** Concurrent processTask threads (redis mode) */ private int consumerThreads = 8; + /** Concurrent threads for monitor order-source sync */ + private int orderSourceSyncThreads = 8; private Oracle oracle = new Oracle(); public String getMqMode() @@ -70,6 +72,16 @@ public class BillingProperties this.consumerThreads = consumerThreads; } + public int getOrderSourceSyncThreads() + { + return orderSourceSyncThreads; + } + + public void setOrderSourceSyncThreads(int orderSourceSyncThreads) + { + this.orderSourceSyncThreads = orderSourceSyncThreads; + } + public Oracle getOracle() { return oracle; diff --git a/bms-back/bms-system/src/main/java/com/bms/price/task/TabOrderSyncAsyncRunner.java b/bms-back/bms-system/src/main/java/com/bms/price/task/TabOrderSyncAsyncRunner.java index 136c3d59..5bf3bdcc 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/task/TabOrderSyncAsyncRunner.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/task/TabOrderSyncAsyncRunner.java @@ -1,7 +1,18 @@ package com.bms.price.task; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; +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.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -10,6 +21,7 @@ import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import com.bms.common.core.redis.RedisCache; import com.bms.common.utils.StringUtils; +import com.bms.price.config.BillingProperties; import com.bms.price.domain.PrcBillingTask; import com.bms.price.domain.PrcBillingTaskRetryTask; import com.bms.price.domain.vo.BillingTaskSubmitRequest; @@ -22,7 +34,7 @@ import com.bms.price.search.TabOrderSyncEsLog; import com.bms.price.service.IPrcBillingTaskService; /** - * Page Oracle TAB_ORDER then submitAndProcess each row. + * Page Oracle TAB_ORDER then multi-thread submitAndProcess each row. */ @Component public class TabOrderSyncAsyncRunner @@ -34,6 +46,8 @@ public class TabOrderSyncAsyncRunner public static final int DEFAULT_BATCH = 50; public static final int MIN_BATCH = 20; public static final int MAX_BATCH = 200; + public static final int MIN_THREADS = 2; + public static final int MAX_THREADS = 16; @Autowired private TabOrderMonitorService tabOrderMonitorService; @@ -48,10 +62,15 @@ public class TabOrderSyncAsyncRunner @Autowired private IBillSearchStore billSearchStore; + @Autowired + private BillingProperties billingProperties; + @Async("threadPoolTaskExecutor") public void run(String asyncTaskId, TabOrderMonitorQuery query, int batchSize) { int batch = clampBatch(batchSize); + int threads = clampThreads(billingProperties == null ? null : billingProperties.getOrderSourceSyncThreads()); + ExecutorService pool = newWorkerPool(threads, "order-source-sync"); try { long totalLong = tabOrderMonitorService.count(query); @@ -62,9 +81,9 @@ public class TabOrderSyncAsyncRunner put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_SUCCESS, 100, "no tasks", 0, 0, 0, 0, null); return; } - int processed = 0; - int ok = 0; - int fail = 0; + AtomicInteger processed = new AtomicInteger(0); + AtomicInteger ok = new AtomicInteger(0); + AtomicInteger fail = new AtomicInteger(0); int pageNum = 1; while (true) { @@ -74,35 +93,34 @@ public class TabOrderSyncAsyncRunner break; } put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, - Math.min(99, (int) ((processed * 100L) / Math.max(1, total))), - "processing " + processed + "/" + total, total, processed, ok, fail, null); + Math.min(99, (int) ((processed.get() * 100L) / Math.max(1, total))), + "processing " + processed.get() + "/" + total, + total, processed.get(), ok.get(), fail.get(), null); + + List> jobs = new ArrayList>(rows.size()); for (TabOrderMonitorVo row : rows) { - try - { - PrcBillingTask task = billingTaskService.submitAndProcess(toSubmit(row)); - if (task != null && BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus())) + final TabOrderMonitorVo one = row; + jobs.add(() -> { + boolean success = syncOne(one); + if (success) { - ok++; + ok.incrementAndGet(); } else { - fail++; + fail.incrementAndGet(); } - } - catch (RuntimeException e) - { - fail++; - log.warn("orderSource sync failed orderId={} transfer={} msg={}", - row == null ? null : row.getOrderId(), - row == null ? null : row.getTransferOrderId(), - e.getMessage()); - } - processed++; - int percent = Math.min(99, (int) ((processed * 100L) / Math.max(1, total))); - put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, percent, - "processing " + processed + "/" + total, total, processed, ok, fail, null); + int done = processed.incrementAndGet(); + int percent = Math.min(99, (int) ((done * 100L) / Math.max(1, total))); + put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, percent, + "processing " + done + "/" + total, + total, done, ok.get(), fail.get(), null); + return Boolean.valueOf(success); + }); } + invokeAllQuiet(pool, jobs); + if (rows.size() < batch) { break; @@ -110,15 +128,18 @@ public class TabOrderSyncAsyncRunner pageNum++; } put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_SUCCESS, 100, - "done ok=" + ok + " fail=" + fail, total, processed, ok, fail, null); - java.util.Map resp = new java.util.LinkedHashMap(); + "done ok=" + ok.get() + " fail=" + fail.get(), + total, processed.get(), ok.get(), fail.get(), null); + Map resp = new LinkedHashMap(); resp.put("asyncTaskId", asyncTaskId); resp.put("total", Integer.valueOf(total)); - resp.put("processed", Integer.valueOf(processed)); - resp.put("ok", Integer.valueOf(ok)); - resp.put("fail", Integer.valueOf(fail)); + resp.put("processed", Integer.valueOf(processed.get())); + resp.put("ok", Integer.valueOf(ok.get())); + resp.put("fail", Integer.valueOf(fail.get())); + resp.put("threads", Integer.valueOf(threads)); TabOrderSyncEsLog.save(billSearchStore, asyncTaskId, "FILTER_DONE", query, resp, "SUCCESS", null); - log.info("orderSource sync done asyncTaskId={} total={} ok={} fail={}", asyncTaskId, total, ok, fail); + log.info("orderSource sync done asyncTaskId={} total={} ok={} fail={} threads={}", + asyncTaskId, total, ok.get(), fail.get(), threads); } catch (Exception e) { @@ -131,6 +152,159 @@ public class TabOrderSyncAsyncRunner put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_FAIL, 100, "failed", 0, 0, 0, 0, msg); TabOrderSyncEsLog.save(billSearchStore, asyncTaskId, "FILTER_DONE", query, null, "FAIL", msg); } + finally + { + shutdownPool(pool); + } + } + + /** Sync selected rows with a short-lived worker pool (for HTTP syncSelected). */ + public Map syncRowsParallel(List rows) + { + Map data = new HashMap(); + List> details = new ArrayList>(); + if (rows == null || rows.isEmpty()) + { + data.put("ok", Integer.valueOf(0)); + data.put("fail", Integer.valueOf(0)); + data.put("details", details); + return data; + } + int threads = Math.min(clampThreads(billingProperties == null ? null : billingProperties.getOrderSourceSyncThreads()), + Math.max(1, rows.size())); + ExecutorService pool = newWorkerPool(threads, "order-source-sel"); + AtomicInteger ok = new AtomicInteger(0); + AtomicInteger fail = new AtomicInteger(0); + try + { + List>> jobs = new ArrayList>>(rows.size()); + for (TabOrderMonitorVo row : rows) + { + final TabOrderMonitorVo one = row; + jobs.add(() -> { + Map item = new HashMap(); + item.put("orderId", one == null ? null : one.getOrderId()); + item.put("transferOrderId", one == null ? null : one.getTransferOrderId()); + try + { + PrcBillingTask task = billingTaskService.submitAndProcess(toSubmit(one)); + item.put("successStatus", task == null ? null : task.getSuccessStatus()); + item.put("errorMessage", task == null ? null : task.getErrorMessage()); + if (task != null && BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus())) + { + ok.incrementAndGet(); + } + else + { + fail.incrementAndGet(); + } + } + catch (RuntimeException e) + { + fail.incrementAndGet(); + item.put("successStatus", "ERROR"); + item.put("errorMessage", e.getMessage()); + } + return item; + }); + } + List>> futures = pool.invokeAll(jobs); + for (Future> f : futures) + { + try + { + details.add(f.get()); + } + catch (ExecutionException e) + { + fail.incrementAndGet(); + Map item = new HashMap(); + item.put("successStatus", "ERROR"); + item.put("errorMessage", e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); + details.add(item); + } + } + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + throw new com.bms.common.exception.ServiceException("order source sync interrupted"); + } + finally + { + shutdownPool(pool); + } + data.put("ok", Integer.valueOf(ok.get())); + data.put("fail", Integer.valueOf(fail.get())); + data.put("details", details); + data.put("threads", Integer.valueOf(threads)); + return data; + } + + private boolean syncOne(TabOrderMonitorVo row) + { + try + { + PrcBillingTask task = billingTaskService.submitAndProcess(toSubmit(row)); + return task != null && BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus()); + } + catch (RuntimeException e) + { + log.warn("orderSource sync failed orderId={} transfer={} msg={}", + row == null ? null : row.getOrderId(), + row == null ? null : row.getTransferOrderId(), + e.getMessage()); + return false; + } + } + + private static void invokeAllQuiet(ExecutorService pool, List> jobs) throws InterruptedException + { + List> futures = pool.invokeAll(jobs); + for (Future f : futures) + { + try + { + f.get(); + } + catch (ExecutionException e) + { + // counted in job body + } + } + } + + private static ExecutorService newWorkerPool(int threads, String namePrefix) + { + AtomicInteger seq = new AtomicInteger(0); + ThreadFactory factory = r -> { + Thread t = new Thread(r); + t.setName(namePrefix + "-" + seq.incrementAndGet()); + t.setDaemon(true); + return t; + }; + return Executors.newFixedThreadPool(Math.max(1, threads), factory); + } + + private static void shutdownPool(ExecutorService pool) + { + if (pool == null) + { + return; + } + pool.shutdown(); + try + { + if (!pool.awaitTermination(30, TimeUnit.SECONDS)) + { + pool.shutdownNow(); + } + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + pool.shutdownNow(); + } } public void put(String taskId, String status, int percent, String message, @@ -168,6 +342,20 @@ public class TabOrderSyncAsyncRunner return n; } + public static int clampThreads(Integer threads) + { + int n = threads == null || threads.intValue() <= 0 ? 8 : threads.intValue(); + if (n < MIN_THREADS) + { + return MIN_THREADS; + } + if (n > MAX_THREADS) + { + return MAX_THREADS; + } + return n; + } + public static BillingTaskSubmitRequest toSubmit(TabOrderMonitorVo row) { if (row == null)