diff --git a/bms-back/ruoyi-admin/src/main/resources/application.yml b/bms-back/ruoyi-admin/src/main/resources/application.yml index 2aed60c7..a5ebecd7 100644 --- a/bms-back/ruoyi-admin/src/main/resources/application.yml +++ b/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 diff --git a/bms-back/ruoyi-system/src/main/java/com/ruoyi/price/config/BillingProperties.java b/bms-back/ruoyi-system/src/main/java/com/ruoyi/price/config/BillingProperties.java index 80ad4f95..e464ab07 100644 --- a/bms-back/ruoyi-system/src/main/java/com/ruoyi/price/config/BillingProperties.java +++ b/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; diff --git a/bms-back/ruoyi-system/src/main/java/com/ruoyi/price/mq/BillingTaskMqConsumer.java b/bms-back/ruoyi-system/src/main/java/com/ruoyi/price/mq/BillingTaskMqConsumer.java index b334e679..12906486 100644 --- a/bms-back/ruoyi-system/src/main/java/com/ruoyi/price/mq/BillingTaskMqConsumer.java +++ b/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(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(); diff --git a/bms-front/src/views/price/basic/surcharge/components/SurchargeFormDialog.vue b/bms-front/src/views/price/basic/surcharge/components/SurchargeFormDialog.vue index 41082411..108f87d2 100644 --- a/bms-front/src/views/price/basic/surcharge/components/SurchargeFormDialog.vue +++ b/bms-front/src/views/price/basic/surcharge/components/SurchargeFormDialog.vue @@ -117,7 +117,7 @@ function emptyForm() { scName: undefined, feeType: 'SURCHARGE', chargeType: 'FIXED', - chargeValue: undefined, + chargeValue: 0, currency: 'CNY', roundMode: 'HALF_UP', countryCode: undefined,