统一修改ruoyi->bms

2026-09-28 10:02:38 +08:00
parent 341afc4111
commit 0408b6c35e
639 changed files with 2776 additions and 2807 deletions
@@ -0,0 +1,30 @@
+package com.bms.price.mq;
+
+import java.io.Serializable;
+
+/** MQ payload: only taskId; consumer loads full row from MySQL. */
+public class BillingTaskMessage implements Serializable
+{
+    private static final long serialVersionUID = 1L;
+
+    private Long taskId;
+
+    public BillingTaskMessage()
+    {
+    }
+
+    public BillingTaskMessage(Long taskId)
+    {
+        this.taskId = taskId;
+    }
+
+    public Long getTaskId()
+    {
+        return taskId;
+    }
+
+    public void setTaskId(Long taskId)
+    {
+        this.taskId = taskId;
+    }
+}
@@ -0,0 +1,173 @@
+package com.bms.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;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.event.EventListener;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import com.alibaba.fastjson2.JSON;
+import com.bms.common.utils.StringUtils;
+import com.bms.price.config.BillingProperties;
+import com.bms.price.service.IPrcBillingTaskService;
+
+/**
+ * Consume billing task MQ messages and call billing interface.
+ */
+@Component
+@ConditionalOnProperty(prefix = "bms.billing", name = "consumer-enabled", havingValue = "true", matchIfMissing = true)
+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;
+
+    @Autowired
+    private BillingProperties billingProperties;
+
+    @Autowired(required = false)
+    private StringRedisTemplate stringRedisTemplate;
+
+    /** Local MQ mode: async event listener. */
+    @Async
+    @EventListener
+    public void onLocalEvent(BillingTaskMqEvent event)
+    {
+        if (!isMode("local"))
+        {
+            return;
+        }
+        if (event == null || event.getMessage() == null)
+        {
+            return;
+        }
+        billingTaskService.processTask(event.getMessage().getTaskId());
+    }
+
+    /**
+     * 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"))
+        {
+            return;
+        }
+        if (stringRedisTemplate == null)
+        {
+            return;
+        }
+        String key = billingProperties.getMqQueueKey();
+        int batch = Math.max(1, billingProperties.getConsumerBatchSize());
+        ExecutorService pool = workerPool();
+        int submitted = 0;
+        for (int i = 0; i < batch; i++)
+        {
+            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);
+            if (msg == null || msg.getTaskId() == null)
+            {
+                log.warn("invalid billing mq payload={}", payload);
+                return;
+            }
+            billingTaskService.processTask(msg.getTaskId());
+        }
+        catch (RuntimeException e)
+        {
+            log.error("billing mq consume failed payload={}", payload, e);
+        }
+    }
+
+    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();
+        if (StringUtils.isEmpty(mode))
+        {
+            mode = "redis";
+        }
+        return expected.equalsIgnoreCase(mode.trim());
+    }
+}
@@ -0,0 +1,21 @@
+package com.bms.price.mq;
+
+import org.springframework.context.ApplicationEvent;
+
+public class BillingTaskMqEvent extends ApplicationEvent
+{
+    private static final long serialVersionUID = 1L;
+
+    private final BillingTaskMessage message;
+
+    public BillingTaskMqEvent(Object source, BillingTaskMessage message)
+    {
+        super(source);
+        this.message = message;
+    }
+
+    public BillingTaskMessage getMessage()
+    {
+        return message;
+    }
+}
@@ -0,0 +1,7 @@
+package com.bms.price.mq;
+
+/** Publish billing task id to MQ after warehouse inbound insert. */
+public interface IBillingTaskMqPublisher
+{
+    void publish(BillingTaskMessage message);
+}
@@ -0,0 +1,27 @@
+package com.bms.price.mq;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.stereotype.Component;
+
+/**
+ * Local MQ: Spring application event (dev / no Redis broker).
+ * Enable with bms.billing.mq-mode=local
+ */
+@Component
+@ConditionalOnProperty(prefix = "bms.billing", name = "mq-mode", havingValue = "local")
+public class LocalBillingTaskMqPublisher implements IBillingTaskMqPublisher
+{
+    private final ApplicationEventPublisher eventPublisher;
+
+    public LocalBillingTaskMqPublisher(ApplicationEventPublisher eventPublisher)
+    {
+        this.eventPublisher = eventPublisher;
+    }
+
+    @Override
+    public void publish(BillingTaskMessage message)
+    {
+        eventPublisher.publishEvent(new BillingTaskMqEvent(this, message));
+    }
+}
@@ -0,0 +1,38 @@
+package com.bms.price.mq;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+import com.alibaba.fastjson2.JSON;
+import com.bms.price.config.BillingProperties;
+
+/**
+ * Redis list as MQ queue (default). LPUSH on publish; consumer BRPOP/RPOP.
+ * Enable with bms.billing.mq-mode=redis (default).
+ */
+@Component
+@ConditionalOnProperty(prefix = "bms.billing", name = "mq-mode", havingValue = "redis", matchIfMissing = true)
+public class RedisBillingTaskMqPublisher implements IBillingTaskMqPublisher
+{
+    private static final Logger log = LoggerFactory.getLogger(RedisBillingTaskMqPublisher.class);
+
+    private final StringRedisTemplate stringRedisTemplate;
+    private final BillingProperties billingProperties;
+
+    public RedisBillingTaskMqPublisher(StringRedisTemplate stringRedisTemplate, BillingProperties billingProperties)
+    {
+        this.stringRedisTemplate = stringRedisTemplate;
+        this.billingProperties = billingProperties;
+    }
+
+    @Override
+    public void publish(BillingTaskMessage message)
+    {
+        String key = billingProperties.getMqQueueKey();
+        String payload = JSON.toJSONString(message);
+        stringRedisTemplate.opsForList().leftPush(key, payload);
+        log.info("billing task mq published taskId={} queue={}", message.getTaskId(), key);
+    }
+}