7、Spring Boot 3.x集成RabbitMQ动态实例等操作

一、前言

本篇主要是围绕着 Spring Boot 3.x 与 RabbitMQ 的动态配置集成，对比上一篇文章，进一步集成RabbitMQ动态操作，比如动态新增 RabbitMQ 实例，以及动态实例中的交换机、队列等操作。

二、动态RabbitMQ实例，创建、删除

1、RabbitMQ动态实例配置

DynamicRabbitMQConfig.java

mport com.chain.air.rpp.exchange.properties.RabbitInstance;
import com.chain.air.rpp.exchange.properties.RabbitProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @className: DynamicRabbitMQConfig
 * @program: chain
 * @description: 动态RabbitMQ配置
 * @author: kenny
 * @create: 2024-10-09 22:40
 * @version: 1.0.0
 */
@Slf4j
@Configuration
public class DynamicRabbitMQConfig {

    /**
     * RabbitMQ实例配置信息集合
     */
    private final RabbitProperties rabbitProperties;
    /**
     * RabbitMQ动态实例列表
     */
    private Map<String, RabbitInstance> rabbitInstanceMap = new HashMap<>();
    /**
     * RabbitTemplate集合
     */
    private Map<String, RabbitTemplate> rabbitTemplateMap = new HashMap<>();
    /**
     * RabbitAdmin集合
     */
    private Map<String, RabbitAdmin> rabbitAdminMap = new HashMap<>();
    /**
     * RabbitAdmin集合
     */
    private Map<String, ConnectionFactory> connectionFactoryMap = new HashMap<>();

    @Autowired
    public DynamicRabbitMQConfig(RabbitProperties rabbitProperties) {
        this.rabbitProperties = rabbitProperties;
    }

    /**
     * 初始化 RabbitMQ 实例
     *
     * @PostConstruct 注解，在 Spring 容器初始化后执行
     */
    @PostConstruct
    public void init() {
        rabbitProperties.getInstances().forEach(this::createRabbitInstance);
    }

    /**
     * 创建一个 RabbitMQ 实例
     *
     * @param instance 实例配置信息
     */
    public void createRabbitInstance(RabbitInstance instance) {
        // 创建连接工厂
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(instance.getHost());
        connectionFactory.setPort(instance.getPort());
        connectionFactory.setUsername(instance.getUsername());
        connectionFactory.setPassword(instance.getPassword());
        connectionFactory.setVirtualHost(instance.getVirtualHost());


        connectionFactoryMap.put(instance.getName(), connectionFactory);

        // 创建 RabbitTemplate
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置消息转换器
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter(new ObjectMapper()));
        rabbitTemplateMap.put(instance.getName(), rabbitTemplate);

        // 创建 RabbitAdmin
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        rabbitAdminMap.put(instance.getName(), rabbitAdmin);

        // 存储实例信息
        rabbitInstanceMap.put(instance.getName(), instance);
    }

    /**
     * 根据名称获取 RabbitTemplate
     *
     * @param name 名称
     * @return RabbitTemplate
     */
    public RabbitTemplate getRabbitTemplate(String name) {
        return rabbitTemplateMap.get(name);
    }

    /**
     * 根据名称获取 RabbitAdmin
     *
     * @param name 名称
     * @return RabbitAdmin
     */
    public RabbitAdmin getRabbitAdmin(String name) {
        return rabbitAdminMap.get(name);
    }

    /**
     * 根据名称获取 ConnectionFactory
     *
     * @param name 名称
     * @return ConnectionFactory
     */
    public ConnectionFactory getConnectionFactory(String name) {
        return connectionFactoryMap.get(name);
    }

    /**
     * 根据名称获取 RabbitTemplate 和 RabbitAdmin
     *
     * @param name 名称
     * @return Map
     */
    public Map<String, Object> getRabbitInstance(String name) {
        Map<String, Object> result = new HashMap<>();
        result.put("instance", rabbitInstanceMap.get(name));
        result.put("rabbitAdmin", rabbitAdminMap.get(name));
        result.put("rabbitTemplate", rabbitTemplateMap.get(name));
        result.put("connectionFactory", connectionFactoryMap.get(name));
        return result;
    }

    /**
     * 检查 RabbitMQ 实例是否存在
     *
     * @param name 名称
     * @return Boolean
     */
    public Boolean checkInstanceExist(String name) {
        return rabbitInstanceMap.containsKey(name);
    }

    /**
     * 获取RabbitMQ动态实例名称集合
     *
     * @return RabbitMQ动态实例名称集合
     */
    public List<String> getDynamicInstanceNames() {
        return new ArrayList<>(rabbitInstanceMap.keySet());
    }

    /**
     * 根据名称删除RabbitMQ实例
     *
     * @param name 名称
     */
    public void removeRabbitInstance(String name) {
        rabbitInstanceMap.remove(name);
        rabbitAdminMap.remove(name);
        rabbitTemplateMap.remove(name);
        connectionFactoryMap.remove(name);
    }
}

2、RabbitMQ动态实例操作Service组件

RabbitDynamicInstanceService.java

import com.chain.air.rpp.exchange.config.rabbit.DynamicRabbitMQConfig;
import com.chain.air.rpp.exchange.properties.RabbitInstance;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @className: RabbitDynamicInstanceService
 * @program: chain
 * @description: RabbitMQ 动态实例 Service组件
 * @author: kenny
 * @create: 2024-10-10 21:56
 * @version: 1.0.0
 */
@Slf4j
@Service
public class RabbitDynamicInstanceService {

    /**
     * RabbitMQ动态实例配置
     */
    private final DynamicRabbitMQConfig dynamicRabbitMQConfig;

    @Autowired
    public RabbitDynamicInstanceService(DynamicRabbitMQConfig dynamicRabbitMQConfig) {
        this.dynamicRabbitMQConfig = dynamicRabbitMQConfig;
    }

    /**
     * 创建RabbitMQ动态实例
     *
     * @param rabbitInstance 实例配置信息
     * @return 处理结果
     */
    public Boolean createRabbitInstance(RabbitInstance rabbitInstance) {
        try {
            boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(rabbitInstance.getName());
            if (instanceExist) {
                log.warn("实例【{}】已存在，无需重复创建", rabbitInstance.getName());
                return true;
            }

            dynamicRabbitMQConfig.createRabbitInstance(rabbitInstance);
            return true;
        } catch (Exception e) {
            log.error("创建RabbitMQ实例失败，失败原因：{}", e.getMessage());
            return false;
        }
    }

    /**
     * 删除RabbitMQ动态实例
     *
     * @param name 实例名称
     * @return 处理结果
     */
    public Boolean removeRabbitInstance(String name) {
        try {
            boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(name);
            if (!instanceExist) {
                log.warn("实例【{}】不存在，无需删除", name);
                return true;
            }

            dynamicRabbitMQConfig.removeRabbitInstance(name);
            return true;
        } catch (Exception e) {
            log.error("删除RabbitMQ实例失败，失败原因：{}", e.getMessage());
            return false;
        }
    }

    /**
     * 获取RabbitMQ动态实例名称列表
     *
     * @return 实例名称列表
     */
    public List<String> getDynamicInstanceNames() {
        return dynamicRabbitMQConfig.getDynamicInstanceNames();
    }

    /**
     * 指定动态实例，通过exchange发送消息
     *
     * @param name       实例名称
     * @param exchange   交换机
     * @param routingKey 路由键
     * @param message    消息内容
     * @return 处理结果
     */
    public Boolean appointRabbitWithExchangeSendMessage(String name, String exchange, String routingKey, String message) {
        try {
            Boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(name);
            if (!instanceExist) {
                log.warn("实例【{}】不存在，无法发送消息", name);
                return false;
            }

            dynamicRabbitMQConfig.sendMessage(name, exchange, routingKey, message);
            return true;
        } catch (Exception e) {
            log.error("指定RabbitMQ实例发送消息失败，失败原因：{}", e.getMessage());
            return false;
        }
    }

    /**
     * 指定动态实例，通过queue发送消息
     *
     * @param name    实例名称
     * @param queue   消息队列
     * @param message 消息内容
     * @return 处理结果
     */
    public Boolean appointRabbitWithQueueSendMessage(String name, String queue, String message) {
        try {
            Boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(name);
            if (!instanceExist) {
                log.warn("实例【{}】不存在，无法发送消息", name);
                return false;
            }

            dynamicRabbitMQConfig.sendMessage(name, queue, message);
            return true;
        } catch (Exception e) {
            log.error("指定RabbitMQ实例发送消息失败，失败原因：{}", e.getMessage());
            return false;
        }
    }

}

二、动态RabbitMQ实例中的exchange、queue动态新增及监听

1、DynamicRabbitMQConfig.java 新增操作代码

import com.chain.air.rpp.exchange.properties.RabbitInstance;
import com.chain.air.rpp.exchange.properties.RabbitProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @className: DynamicRabbitMQConfig
 * @program: chain
 * @description: 动态RabbitMQ配置
 * @author: kenny
 * @create: 2024-10-09 22:40
 * @version: 1.0.0
 */
@Slf4j
@Configuration
public class DynamicRabbitMQConfig {

    /**
     * RabbitMQ实例配置信息集合
     */
    private final RabbitProperties rabbitProperties;
    /**
     * RabbitMQ动态实例列表
     */
    private Map<String, RabbitInstance> rabbitInstanceMap = new HashMap<>();
    /**
     * RabbitTemplate集合
     */
    private Map<String, RabbitTemplate> rabbitTemplateMap = new HashMap<>();
    /**
     * RabbitAdmin集合
     */
    private Map<String, RabbitAdmin> rabbitAdminMap = new HashMap<>();
    /**
     * RabbitAdmin集合
     */
    private Map<String, ConnectionFactory> connectionFactoryMap = new HashMap<>();

    @Autowired
    public DynamicRabbitMQConfig(RabbitProperties rabbitProperties) {
        this.rabbitProperties = rabbitProperties;
    }

    /**
     * 初始化 RabbitMQ 实例
     *
     * @PostConstruct 注解，在 Spring 容器初始化后执行
     */
    @PostConstruct
    public void init() {
        rabbitProperties.getInstances().forEach(this::createRabbitInstance);
    }

    /**
     * 创建一个 RabbitMQ 实例
     *
     * @param instance 实例配置信息
     */
    public void createRabbitInstance(RabbitInstance instance) {
        // 创建连接工厂
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(instance.getHost());
        connectionFactory.setPort(instance.getPort());
        connectionFactory.setUsername(instance.getUsername());
        connectionFactory.setPassword(instance.getPassword());
        connectionFactory.setVirtualHost(instance.getVirtualHost());


        connectionFactoryMap.put(instance.getName(), connectionFactory);

        // 创建 RabbitTemplate
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置消息转换器
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter(new ObjectMapper()));
        rabbitTemplateMap.put(instance.getName(), rabbitTemplate);

        // 创建 RabbitAdmin
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        rabbitAdminMap.put(instance.getName(), rabbitAdmin);

        // 存储实例信息
        rabbitInstanceMap.put(instance.getName(), instance);
    }

    /**
     * 根据名称获取 RabbitTemplate
     *
     * @param name 名称
     * @return RabbitTemplate
     */
    public RabbitTemplate getRabbitTemplate(String name) {
        return rabbitTemplateMap.get(name);
    }

    /**
     * 根据名称获取 RabbitAdmin
     *
     * @param name 名称
     * @return RabbitAdmin
     */
    public RabbitAdmin getRabbitAdmin(String name) {
        return rabbitAdminMap.get(name);
    }

    /**
     * 根据名称获取 ConnectionFactory
     *
     * @param name 名称
     * @return ConnectionFactory
     */
    public ConnectionFactory getConnectionFactory(String name) {
        return connectionFactoryMap.get(name);
    }

    /**
     * 根据名称获取 RabbitTemplate 和 RabbitAdmin
     *
     * @param name 名称
     * @return Map
     */
    public Map<String, Object> getRabbitInstance(String name) {
        Map<String, Object> result = new HashMap<>();
        result.put("instance", rabbitInstanceMap.get(name));
        result.put("rabbitAdmin", rabbitAdminMap.get(name));
        result.put("rabbitTemplate", rabbitTemplateMap.get(name));
        result.put("connectionFactory", connectionFactoryMap.get(name));
        return result;
    }

    /**
     * 检查 RabbitMQ 实例是否存在
     *
     * @param name 名称
     * @return Boolean
     */
    public Boolean checkInstanceExist(String name) {
        return rabbitInstanceMap.containsKey(name);
    }

    /**
     * 获取RabbitMQ动态实例名称集合
     *
     * @return RabbitMQ动态实例名称集合
     */
    public List<String> getDynamicInstanceNames() {
        return new ArrayList<>(rabbitInstanceMap.keySet());
    }

    /**
     * 根据名称删除RabbitMQ实例
     *
     * @param name 名称
     */
    public void removeRabbitInstance(String name) {
        rabbitInstanceMap.remove(name);
        rabbitAdminMap.remove(name);
        rabbitTemplateMap.remove(name);
        connectionFactoryMap.remove(name);
    }

    /**
     * 根据name名称往指定 exchange 发送消息
     *
     * @param name       名称
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    消息内容
     */
    public void sendMessage(String name, String exchange, String routingKey, String message) {
        try {
            RabbitTemplate rabbitTemplate = rabbitTemplateMap.get(name);
            rabbitTemplate.convertAndSend(exchange, routingKey, message);
        } catch (Exception e) {
            log.error("RabbitMQ实例对象：{}，发送消息失败，失败原因：{}", name, e.getMessage());
            log.error("异常堆栈信息：{}", e);
        }
    }

    /**
     * 根据name名称往指定 queue 发送消息
     *
     * @param name    名称
     * @param queue   队列名称
     * @param message 消息内容
     */
    public void sendMessage(String name, String queue, String message) {
        RabbitTemplate rabbitTemplate = rabbitTemplateMap.get(name);
        rabbitTemplate.convertAndSend(queue, message);
    }

    // 根据name名称获取rabbitAdmin，创建Direct交换机
    public void createDirectExchange(String name, String exchangeName) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);

        DirectExchange directExchange = new DirectExchange(exchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        log.info("RabbitMQ实例对象：{}，Direct 交换机创建成功: {}", name, exchangeName);
    }

    // 根据name名称获取rabbitAdmin，创建Fount交换机
    public void createFountExchange(String name, String exchangeName) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);
        DirectExchange fanoutExchange = new DirectExchange(exchangeName, true, false);
        rabbitAdmin.declareExchange(fanoutExchange);
        log.info("RabbitMQ实例对象：{}，Fanout 交换机创建成功: {}", name, exchangeName);
    }

    // 根据name名称获取rabbitAdmin，创建Topic交换机
    public void createTopicExchange(String name, String exchangeName) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);
        TopicExchange topicExchange = new TopicExchange(exchangeName, true, false);
        rabbitAdmin.declareExchange(topicExchange);
        log.info("RabbitMQ实例对象：{}，Topic 交换机创建成功: {}", name, exchangeName);
    }

    // 根据name名称获取rabbitAdmin，创建Headers交换机
    public void createHeadersExchange(String name, String exchangeName) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);
        HeadersExchange headersExchange = new HeadersExchange(exchangeName, true, false);
        rabbitAdmin.declareExchange(headersExchange);
        log.info("RabbitMQ实例对象：{}，Headers 交换机创建成功: {}", name, exchangeName);
    }

    // 根据name名称获取rabbitAdmin，创建队列
    public void createQueue(String name, String queueName) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);
        Queue queue = new Queue(queueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        log.info("RabbitMQ实例对象：{}，队列创建成功: {}", name, queueName);
    }

    // 根据name名称获取rabbitAdmin，删除队列
    public void deleteQueue(String name, String queueName) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);
        rabbitAdmin.deleteQueue(queueName);
        log.info("RabbitMQ实例对象：{}，队列删除成功: {}", name, queueName);
    }

    // 根据name名称获取rabbitAdmin，创建队列并监听
    public void createQueue(String name, String queueName, Boolean isListener) {
        RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);
        ConnectionFactory connectionFactory = connectionFactoryMap.get(name);
        Queue queue = new Queue(queueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        log.info("RabbitMQ实例对象：{}，队列创建成功: {}", name, queueName);

        if (isListener) {
            createListener(connectionFactory, queueName);
        }
    }

    /**
     * 创建监听器容器并启动监听
     *
     * @param queueName 队列名称
     */
    public void createListener(ConnectionFactory connectionFactory, String queueName) {
        // 创建并启动监听器容器
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);

        // 监听逻辑处理
        container.setMessageListener(new MessageListenerAdapter(new Object() {
            public void handleMessage(String message) {
                System.out.println("收到来自RabbitMQ中队列：" + queueName + " 队列的消息：" + message);
            }
        }));

        // 启动监听器容器
        container.start();
        System.out.println("RabbitMQ队列监听器已启动：" + queueName);
    }
}