Quellcode durchsuchen

添加判断查看初始化的队列是否存在,不存在则新建

cxd vor 8 Monaten
Ursprung
Commit
9d95901d6e

+ 109 - 0
module-erp/src/main/java/com/hys/app/model/base/rabbitmq/RabbitMQConfig.java

@@ -0,0 +1,109 @@
+package com.hys.app.model.base.rabbitmq;
+
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.*;
+
+@Configuration
+@Slf4j
+public class RabbitMQConfig {
+
+    @Autowired
+    private AmqpAdmin amqpAdmin;
+
+    private final List<String> exchangeNames = new ArrayList<String>() {{
+        add("STOCK_DAMAGE_AUDIT_PASS");
+        add("ERP_GOODS_CHANGE");
+        add("ORDER_RETURN_AUDIT_PASS");
+        add("STOCK_STATISTICS");
+        add("LOGS");
+        add("WAREHOUSE_ENTRY_AUDIT_PASS");
+        add("WAREHOUSE_OUT_SHIP");
+    }};
+
+    @PostConstruct
+    public void init() {
+        for (String exchangeName : exchangeNames) {
+            try {
+                String queueName = exchangeName + "_QUEUE";
+                initializeIfNotExists(exchangeName, queueName);
+            } catch (Exception e) {
+                log.error("Failed to initialize exchange '{}': {}", exchangeName, e.getMessage());
+                log.error("Error details:", e);
+                throw new RuntimeException("Failed to initialize RabbitMQ resources: " + e.getMessage(), e);
+            }
+        }
+    }
+
+    private void initializeIfNotExists(String exchangeName, String queueName) {
+        try {
+            // 检查队列是否存在
+            Properties queueProperties = amqpAdmin.getQueueProperties(queueName);
+            if (queueProperties != null) {
+                log.info("Queue '{}' already exists, skipping creation", queueName);
+                return; // 队列存在,直接返回
+            }
+
+            log.info("Queue '{}' does not exist, creating new...", queueName);
+            initializeQueue(exchangeName);
+
+        } catch (Exception e) {
+            log.warn("Error checking queue existence for '{}': {}", queueName, e.getMessage());
+            // 如果是队列不存在导致的异常,则创建队列
+            if (e.getCause() instanceof com.rabbitmq.client.ShutdownSignalException) {
+                log.info("Attempting to create queue '{}'", queueName);
+                initializeQueue(exchangeName);
+            } else {
+                throw e; // 其他异常则抛出
+            }
+        }
+    }
+
+    private void initializeQueue(String exchangeName) {
+        String queueName = exchangeName + "_QUEUE";
+
+        try {
+            // 1. 创建交换机
+            Exchange exchange = ExchangeBuilder
+                    .fanoutExchange(exchangeName)
+                    .durable(true)
+                    .build();
+            amqpAdmin.declareExchange(exchange);
+            log.info("Created fanout exchange '{}'", exchangeName);
+
+            // 2. 创建队列
+            Queue queue = QueueBuilder
+                    .durable(queueName)
+                    .build();
+            amqpAdmin.declareQueue(queue);
+            log.info("Created queue '{}'", queueName);
+
+            // 3. 创建绑定
+            Binding binding = BindingBuilder
+                    .bind(queue)
+                    .to(exchange)
+                    .with("")
+                    .noargs();
+            amqpAdmin.declareBinding(binding);
+            log.info("Created binding between '{}' and '{}'", exchangeName, queueName);
+
+        } catch (Exception e) {
+            log.error("Failed to initialize queue '{}': {}", queueName, e.getMessage());
+            if (e.getCause() != null) {
+                log.error("Caused by: ", e.getCause());
+            }
+            throw e;
+        }
+    }
+
+}