修改多线程批量插入

bms-back/ruoyi-admin/src/main/resources/application.yml

@@ -156,7 +156,11 @@ bms:
     mq-mode: redis
     mq-queue-key: bms:billing:task
     consumer-enabled: true
-    consumer-delay-ms: 1000
+    # Pop/process interval after previous tick finishes
+    consumer-delay-ms: 200
+    # Max messages popped per tick; processed by consumer-threads pool
+    consumer-batch-size: 32
+    consumer-threads: 8
     oracle:
       # SLAVE datasource Oracle main-order table (schema.table allowed)
       order-table: WEISHI_USER.TAB_ORDER

bms-back/ruoyi-system/src/main/java/com/ruoyi/price/config/BillingProperties.java

@@ -14,6 +14,10 @@ public class BillingProperties
     private String mqMode = "redis";
     private String mqQueueKey = "bms:billing:task";
     private boolean consumerEnabled = true;
+    /** Messages popped per scheduled tick (redis mode) */
+    private int consumerBatchSize = 32;
+    /** Concurrent processTask threads (redis mode) */
+    private int consumerThreads = 8;
     private Oracle oracle = new Oracle();
 
     public String getMqMode()
@@ -46,6 +50,26 @@ public class BillingProperties
         this.consumerEnabled = consumerEnabled;
     }
 
+    public int getConsumerBatchSize()
+    {
+        return consumerBatchSize;
+    }
+
+    public void setConsumerBatchSize(int consumerBatchSize)
+    {
+        this.consumerBatchSize = consumerBatchSize;
+    }
+
+    public int getConsumerThreads()
+    {
+        return consumerThreads;
+    }
+
+    public void setConsumerThreads(int consumerThreads)
+    {
+        this.consumerThreads = consumerThreads;
+    }
+
     public Oracle getOracle()
     {
         return oracle;

bms-back/ruoyi-system/src/main/java/com/ruoyi/price/mq/BillingTaskMqConsumer.java

@@ -1,5 +1,11 @@
 package com.ruoyi.price.mq;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import jakarta.annotation.PreDestroy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -23,6 +29,9 @@ public class BillingTaskMqConsumer
 {
     private static final Logger log = LoggerFactory.getLogger(BillingTaskMqConsumer.class);
 
+    private final AtomicInteger workerSeq = new AtomicInteger();
+    private volatile ExecutorService workerPool;
+
     @Autowired
     private IPrcBillingTaskService billingTaskService;
 
@@ -48,8 +57,11 @@ public class BillingTaskMqConsumer
         billingTaskService.processTask(event.getMessage().getTaskId());
     }
 
-    /** Redis MQ mode: poll list every second. */
-    @Scheduled(fixedDelayString = "${bms.billing.consumer-delay-ms:1000}")
+    /**
+     * Redis MQ mode: pop a batch each tick and process concurrently.
+     * Previously only 1 message / schedule cycle — queue backlog grew while inserts were fast.
+     */
+    @Scheduled(fixedDelayString = "${bms.billing.consumer-delay-ms:200}")
     public void pollRedisQueue()
     {
         if (!isMode("redis"))
@@ -61,11 +73,28 @@ public class BillingTaskMqConsumer
             return;
         }
         String key = billingProperties.getMqQueueKey();
-        String payload = stringRedisTemplate.opsForList().rightPop(key);
-        if (payload == null || payload.isEmpty())
+        int batch = Math.max(1, billingProperties.getConsumerBatchSize());
+        ExecutorService pool = workerPool();
+        int submitted = 0;
+        for (int i = 0; i < batch; i++)
         {
-            return;
+            String payload = stringRedisTemplate.opsForList().rightPop(key);
+            if (payload == null || payload.isEmpty())
+            {
+                break;
+            }
+            final String body = payload;
+            pool.execute(() -> handlePayload(body));
+            submitted++;
         }
+        if (submitted > 0 && log.isDebugEnabled())
+        {
+            log.debug("billing mq popped batchSize={}", submitted);
+        }
+    }
+
+    private void handlePayload(String payload)
+    {
         try
         {
             BillingTaskMessage msg = JSON.parseObject(payload, BillingTaskMessage.class);
@@ -82,6 +111,56 @@ public class BillingTaskMqConsumer
         }
     }
 
+    private ExecutorService workerPool()
+    {
+        ExecutorService existing = workerPool;
+        if (existing != null)
+        {
+            return existing;
+        }
+        synchronized (this)
+        {
+            if (workerPool == null)
+            {
+                int threads = Math.max(1, billingProperties.getConsumerThreads());
+                AtomicInteger seq = workerSeq;
+                workerPool = new ThreadPoolExecutor(
+                    threads,
+                    threads,
+                    60L,
+                    TimeUnit.SECONDS,
+                    new LinkedBlockingQueue<Runnable>(threads * 4),
+                    r -> {
+                        Thread t = new Thread(r);
+                        t.setName("billing-mq-" + seq.incrementAndGet());
+                        t.setDaemon(true);
+                        return t;
+                    },
+                    new ThreadPoolExecutor.CallerRunsPolicy());
+            }
+            return workerPool;
+        }
+    }
+
+    @PreDestroy
+    public void shutdown()
+    {
+        ExecutorService pool = workerPool;
+        if (pool != null)
+        {
+            pool.shutdown();
+            try
+            {
+                pool.awaitTermination(30, TimeUnit.SECONDS);
+            }
+            catch (InterruptedException e)
+            {
+                Thread.currentThread().interrupt();
+                pool.shutdownNow();
+            }
+        }
+    }
+
     private boolean isMode(String expected)
     {
         String mode = billingProperties.getMqMode();