Browse Source

提交rabbimq初始化队列程序

cxd 8 months ago
parent
commit
74eec08e57

+ 91 - 0
framework/src/main/java/com/hys/app/framework/rabbitmq/RabbitMQConfig.java

@@ -0,0 +1,91 @@
+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 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("ERP_GOODS_CHANGE");
+        add("CHANGE_FORM_AUDIT_PASS");
+
+        add("ORDER_RETURN_AUDIT_PASS");
+        add("STOCK_DAMAGE_AUDIT_PASS");
+        add("STOCK_STATISTICS");
+        add("SUPPLIER_RETURN_AUDIT_PASS");
+        add("LOGS");
+        add("WAREHOUSE_ENTRY_AUDIT_PASS");
+
+        add("WAREHOUSE_OUT_SHIP");
+        add("MESSAGE_RECEIVE");
+    }};
+
+    @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);
+            } catch (Exception e) {
+                log.warn("Failed to delete exchange/queue '{}': {}", exchangeName, e.getMessage());
+            }
+        }
+
+        // 重新创建所有交换机和队列
+        for (String exchangeName : exchangeNames) {
+            try {
+                initializeQueue(exchangeName);
+            } catch (Exception e) {
+                log.error("Failed to initialize exchange '{}': {}", exchangeName, e.getMessage());
+                throw new RuntimeException("Failed to initialize RabbitMQ resources", e);
+            }
+        }
+    }
+
+    private void initializeQueue(String exchangeName) {
+
+        String queueName = exchangeName + "_QUEUE";
+
+        try {
+            // 1. 创建简单的 Fanout 交换机
+            FanoutExchange exchange = new FanoutExchange(exchangeName, true, false);
+            amqpAdmin.declareExchange(exchange);
+            log.info("Created fanout exchange '{}'", exchangeName);
+
+            // 2. 创建简单的持久化队列
+            Queue queue = new Queue(queueName, true, false, false);
+            amqpAdmin.declareQueue(queue);
+            log.info("Created queue '{}'", queueName);
+
+            // 3. 创建绑定
+            Binding binding = BindingBuilder.bind(queue).to(exchange);
+            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;
+        }
+    }
+}