修改同步速率

2026-10-13 10:33:03 +08:00
parent 4d27887d4a
commit 94b94828ba
4 changed files with 236 additions and 72 deletions
@@ -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;
@@ -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<Callable<Boolean>> jobs = new ArrayList<Callable<Boolean>>(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<String, Object> resp = new java.util.LinkedHashMap<String, Object>();
+                "done ok=" + ok.get() + " fail=" + fail.get(),
+                total, processed.get(), ok.get(), fail.get(), null);
+            Map<String, Object> resp = new LinkedHashMap<String, Object>();
            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<String, Object> syncRowsParallel(List<TabOrderMonitorVo> rows)
+    {
+        Map<String, Object> data = new HashMap<String, Object>();
+        List<Map<String, Object>> details = new ArrayList<Map<String, Object>>();
+        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<Callable<Map<String, Object>>> jobs = new ArrayList<Callable<Map<String, Object>>>(rows.size());
+            for (TabOrderMonitorVo row : rows)
+            {
+                final TabOrderMonitorVo one = row;
+                jobs.add(() -> {
+                    Map<String, Object> item = new HashMap<String, Object>();
+                    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<Future<Map<String, Object>>> futures = pool.invokeAll(jobs);
+            for (Future<Map<String, Object>> f : futures)
+            {
+                try
+                {
+                    details.add(f.get());
+                }
+                catch (ExecutionException e)
+                {
+                    fail.incrementAndGet();
+                    Map<String, Object> item = new HashMap<String, Object>();
+                    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<Callable<Boolean>> jobs) throws InterruptedException
+    {
+        List<Future<Boolean>> futures = pool.invokeAll(jobs);
+        for (Future<Boolean> 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)