当前位置：首页 > article >正文

RabbitMq-队列交换机绑定关系优化为枚举注册

article 2024/10/28 10:23:28

📚目录

📚简介:
🚀比较
- 💨通常注册
- 🌈优化后注册
✍️代码
- 💫自动注册的关键代码

📚简介:

该项目介绍，rabbitMq消息中间件，对队列的注册，交换机的注册，队列与交换机的绑定关系进行注册，这三个步骤进行简化，通过枚举的配置完成这些操作。可以参考之前写的文章>SpringBoot整合Rabbitmq

🚀比较

💨通常注册

通常我们创建一个队列至少三个步骤

声明队列
声明交换机
队列绑定到交换机上，并且且设置路由关键字

在这里插入图片描述

🌈优化后注册

我们只需要关注我们的队列枚举，里面已经定义好了通用配置，只有队列名称，交换机，交换机类型的指定我们需要自己定义外，之后的注册不需要你们关注即可完成。

在这里插入图片描述

✍️代码

💫自动注册的关键代码

RabbitMqConfig

代码的运行逻辑

应用启动：当 Spring Boot 应用启动时，RabbitAdmin Bean 会被初始化并启动。
队列和交换机的创建：在 createQueue 方法中，代码会遍历所有的队列枚举，将每个队列的相关配置（如交换机类型、持久性等）传递给相应的方法，创建所需的队列和交换机。
完成配置：一旦所有的交换机和队列创建完成，它们之间的绑定关系也会被设置好，从而确保消息可以在它们之间正确地流动。

/**
 * @author itmei
 */
@Configuration
public class RabbitMqConfig {

    ///**
    // * 声明队列
    // * @return
    // */
    //@Bean
    //public Queue directQueue(){
    //    //持久 - 如果我们声明一个持久队列，则为真（该队列将在服务器重启后继续存在）
    //    return new Queue(RabbitMqQueueEnum.DEFAULT_DIRECT.getQueueName(),false);
    //}
    //
    ///**
    // * 声明交换机
    // * @return
    // */
    //@Bean
    //public DirectExchange directExchange(){
    //    //交换器名称、是否持久化、是否自动删除
    //    return new DirectExchange(RabbitMqQueueEnum.DEFAULT_DIRECT.getExchangeName(),true,false);
    //}
    //
    ///***
    // * 队列绑定到交换机上
    // * 并且设置路由关键字
    // */
    //@Bean
    //public Binding binding(Queue queue,DirectExchange exchange){
    //    return BindingBuilder.bind(queue).to(exchange).with(RabbitMqQueueEnum.DEFAULT_DIRECT.getRoutingKey());
    //}

    
    /**
     * 用于动态创建队列和交换机
     * @param rabbitTemplate rabbitMq的模板对象
     * @return
     */
    @Bean
    public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        //默认就是true
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    /**
     * 绑定
     * @param rabbitAdmin RabbitAdmin 实例，用于管理队列和交换机的声明
     * @return
     */
    @Bean("createQueue")
    public Object createQueue(RabbitAdmin rabbitAdmin) {
        // 遍历队列枚举
        RabbitMqQueueEnum.toList().forEach(rabbitMqQueueEnum -> {
            //创建交换机
            createExchangeHandle(rabbitAdmin,rabbitMqQueueEnum);
            // 创建对列
            createQueueHandle(rabbitAdmin,rabbitMqQueueEnum);
            // 绑定交换机和对列
            createBindingHandle(rabbitAdmin,rabbitMqQueueEnum);
        });
        return null;
    }


    /**
     * 注册交换机的方法，根据传入的队列枚举配置创建相应类型的交换机。
     * @param rabbitAdmin RabbitAdmin 实例，用于管理队列和交换机的声明
     * @param rabbitMqQueueEnum 自定义队列枚举，包含交换机的配置信息，包含交换机的配置信息
     */
    private void createExchangeHandle(RabbitAdmin rabbitAdmin, RabbitMqQueueEnum rabbitMqQueueEnum) {
        // 获取当前队列的交换机配置
        ExchangeConfigEnum exchangeConfigEnum = rabbitMqQueueEnum.getExchangeConfigEnum();
        String exchangeName = rabbitMqQueueEnum.getExchangeName();

        // 检查是否为延迟交换机
        if (rabbitMqQueueEnum.isDelayedExchange()) {
            // 创建并声明延迟交换机
            rabbitAdmin.declareExchange(new CustomDelayedExchange(
                    // 交换机名称
                    exchangeName,
                    // 交换机是否持久化
                    exchangeConfigEnum.isDurable(),
                    // 交换机是否自动删除
                    exchangeConfigEnum.isAutoDelete(),
                    // 交换机类型
                    exchangeConfigEnum.getType().name().toLowerCase()
            ));
        } else {
            // 创建并声明普通交换机
            createStandardExchange(rabbitAdmin, exchangeName, exchangeConfigEnum);
        }
    }

    /**
     * 创建并声明标准交换机的方法。
     * @param rabbitAdmin RabbitAdmin 实例，用于声明交换机
     * @param exchangeName 交换机的名称
     * @param exchangeConfigEnum 交换机的配置枚举，包含持久化和自动删除等信息
     */
    private void createStandardExchange(RabbitAdmin rabbitAdmin, String exchangeName, ExchangeConfigEnum exchangeConfigEnum) {
        AbstractExchange exchange;

        // 根据交换机类型创建对应的交换机实例
        switch (exchangeConfigEnum.getType()) {
            case FANOUT:
                // 创建 Fanout 交换机
                exchange = new FanoutExchange(exchangeName, exchangeConfigEnum.isDurable(), exchangeConfigEnum.isAutoDelete());
                break;
            case TOPIC:
                // 创建 Topic 交换机
                exchange = new TopicExchange(exchangeName, exchangeConfigEnum.isDurable(), exchangeConfigEnum.isAutoDelete());
                break;
            case DIRECT:
                // 创建 Direct 交换机
                exchange = new DirectExchange(exchangeName, exchangeConfigEnum.isDurable(), exchangeConfigEnum.isAutoDelete());
                break;
            default:
                // 不支持的交换机类型，直接返回
                return;
        }

        // 声明创建的交换机
        rabbitAdmin.declareExchange(exchange);
    }

    /**
     * 注册队列
     * @param rabbitAdmin RabbitAdmin 实例，用于管理队列和交换机的声明
     * @param rabbitMqQueueEnum 自定义队列枚举，包含交换机的配置信息
     */
    public void createQueueHandle(RabbitAdmin rabbitAdmin, RabbitMqQueueEnum rabbitMqQueueEnum) {
        QueueConfigEnum queueEnum = rabbitMqQueueEnum.getQueueConfigEnum();
        rabbitAdmin.declareQueue(new Queue(rabbitMqQueueEnum.getQueueName(),
                queueEnum.isDurable(), queueEnum.isExclusive(), queueEnum.isAutoDelete(), queueEnum.getArguments()));

    }

    /**
     * 注册绑定关系
     * @param rabbitAdmin RabbitAdmin 实例，用于管理队列和交换机的声明
     * @param rabbitMqQueueEnum 自定义队列枚举，包含交换机的配置信息
     */
    public void createBindingHandle(RabbitAdmin rabbitAdmin, RabbitMqQueueEnum rabbitMqQueueEnum) {
        // 绑定
        rabbitAdmin.declareBinding(new Binding(
                // queue名称
                rabbitMqQueueEnum.getQueueName(),
                Binding.DestinationType.QUEUE,
                // exchange名称
                rabbitMqQueueEnum.getExchangeName(),
                // queue的routingKey
                rabbitMqQueueEnum.getRoutingKey(),
                null));
    }

    /**
     * 用于创建延迟队列的交换机
     */
    public static class CustomDelayedExchange extends CustomExchange {
        public CustomDelayedExchange(String name, boolean durable, boolean autoDelete, String delayedType) {
            super(name, "x-delayed-message", durable, autoDelete, Collections.singletonMap("x-delayed-type", delayedType));
        }
    }
}

交换机枚举配置

/**
 * @Author itmei
 */
@Getter
@AllArgsConstructor
public enum ExchangeConfigEnum {
    /**
     * 直接模式
     */
    DIRECT_EXCHANGE(ExchangeModelEnum.DIRECT, true, false),

    /**
     * 匹配模式
     */
    TOPIC_EXCHANGE(ExchangeModelEnum.TOPIC, true, false),

    ;


    /**
     * 模式
     */
    private final ExchangeModelEnum type;

    /**
     * 是否持久化
     */
    private final boolean durable;

    /**
     * 是否自动删除
     */
    private final boolean autoDelete;

}

交换机枚举类型

/**
 * @Author itmei
 */
@Getter
@AllArgsConstructor
public enum ExchangeModelEnum {

    /**
     * 广播
     */
    FANOUT,

    /**
     * 匹配
     */
    TOPIC,

    /**
     * 直接模式
     */
    DIRECT
}

默认队列配置

/**
 * @author itmei
 */
@Getter
@AllArgsConstructor
public enum QueueConfigEnum {
    /**
     * 自定义的队列配置
     */
    DEFAULT_QUEUE_CONFIG(true, false, false, null),
    ;


    /**
     * 是否为持久队列(该队列将在服务器重启后保留下来)
     */
    private final boolean durable;

    /**
     * 是否为排他队列(该队列仅由声明者的队列使用连接)
     */
    private final boolean exclusive;

    /**
     * 如果队列为空是否删除(如果服务器在不再使用队列时是否删除队列)
     */
    private final boolean autoDelete;

    /**
     * queue的参数
     */
    private final Map<String, Object> arguments;


}

自定义队列配置（以后只需要把队列增加到这里即可）


/**
 * @author itmei
 */
@Getter
@AllArgsConstructor
public enum RabbitMqQueueEnum {
    /**
     * 自定义的队列与交换机的关系
     */

    DEFAULT_DIRECT("direct_itmei_exchange",
            "direct.itmei.exchange.default.queue",
            "direct.itmei.exchange.default.queue",
            ExchangeConfigEnum.DIRECT_EXCHANGE,
            QueueConfigEnum.DEFAULT_QUEUE_CONFIG,
            true,
            "默认队列"
    ),
    DEFAULT_TEST_DIRECT("direct_itmei_exchange",
            "direct.itmei.exchange.test.queue",
            "direct.itmei.exchange.test.queue",
            ExchangeConfigEnum.DIRECT_EXCHANGE,
            QueueConfigEnum.DEFAULT_QUEUE_CONFIG,
            true,
            "测试对列"
    ),

    ;

    /**
     * 交换机名称
     */
    private final String exchangeName;

    /**
     * 队列名称(每个队列的名称应是唯一的)
     */
    private final String queueName ;

    /**
     * 默认保持和对列名称一样，模式为 topic时需要注意匹配符设置
     * 队列路由键(
     *  注意匹配，#匹配一个或者多个,*匹配一个
     *      如： abc.new.n , abc.new.n.m
     *      abc.new.# 匹配{abc.new.n,abc.new.n.m}
     *      abc.new.* 匹配{abc.new.n}
     * )
     */
    private final String routingKey;

    /**
     * 交换机的配置
     */
    private final ExchangeConfigEnum exchangeConfigEnum;

    /**
     * 对列的配置
     */
    private final QueueConfigEnum queueConfigEnum;

    /**
     * 是否是延迟交换机 如果已经存在不是延迟交换机则需要删除再标记
     * treu:延迟交换机 false:普通交换机
     */
    private final boolean delayedExchange;

    /**
     * 描述
     */
    private final String desc;

    public static List<RabbitMqQueueEnum> toList() {
        return Arrays.asList(RabbitMqQueueEnum.values());
    }

}