Browse Source

添加mq启动配置

cxd 8 months ago
parent
commit
3ee7d44749

+ 41 - 18
framework/src/main/java/com/hys/app/framework/rabbitmq/RabbitMQConfig.java

@@ -3,13 +3,14 @@ package com.hys.app.framework.rabbitmq;
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.FanoutExchange;
 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.*;
 
@@ -37,46 +38,67 @@ public class RabbitMQConfig {
 
     @PostConstruct
     public void init() {
-        // 先删除所有现有的交换机和队列
         for (String exchangeName : exchangeNames) {
             try {
                 String queueName = exchangeName + "_QUEUE";
-                amqpAdmin.deleteQueue(queueName);
-                amqpAdmin.deleteExchange(exchangeName);
-                log.info("Deleted exchange and queue: {}", exchangeName);
+                initializeIfNotExists(exchangeName, queueName);
             } catch (Exception e) {
-                log.warn("Failed to delete exchange/queue '{}': {}", exchangeName, e.getMessage());
+                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);
             }
         }
+    }
 
-        // 重新创建所有交换机和队列
-        for (String exchangeName : exchangeNames) {
-            try {
+    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);
-            } catch (Exception e) {
-                log.error("Failed to initialize exchange '{}': {}", exchangeName, e.getMessage());
-                throw new RuntimeException("Failed to initialize RabbitMQ resources", e);
+            } else {
+                throw e; // 其他异常则抛出
             }
         }
     }
 
     private void initializeQueue(String exchangeName) {
-
         String queueName = exchangeName + "_QUEUE";
 
         try {
-            // 1. 创建简单的 Fanout 交换机
-            FanoutExchange exchange = new FanoutExchange(exchangeName, true, false);
+            // 1. 创建交换机
+            Exchange exchange = ExchangeBuilder
+                    .fanoutExchange(exchangeName)
+                    .durable(true)
+                    .build();
             amqpAdmin.declareExchange(exchange);
             log.info("Created fanout exchange '{}'", exchangeName);
 
-            // 2. 创建简单的持久化队列
-            Queue queue = new Queue(queueName, true, false, false);
+            // 2. 创建队列
+            Queue queue = QueueBuilder
+                    .durable(queueName)
+                    .build();
             amqpAdmin.declareQueue(queue);
             log.info("Created queue '{}'", queueName);
 
             // 3. 创建绑定
-            Binding binding = BindingBuilder.bind(queue).to(exchange);
+            Binding binding = BindingBuilder
+                    .bind(queue)
+                    .to(exchange)
+                    .with("")
+                    .noargs();
             amqpAdmin.declareBinding(binding);
             log.info("Created binding between '{}' and '{}'", exchangeName, queueName);
 
@@ -88,4 +110,5 @@ public class RabbitMQConfig {
             throw e;
         }
     }
+
 }