RabbitMQ之动态创建队列与绑定交换机和监听器_rabbitmq消费端创建队列并绑定交换机

为什么需要动态创建队列与绑定交换机？我在写项目的时候遇到这么个问题，我数据库中存在一个字段messageType指定为消息类型，消息类型存在三种，一种是通知类，一种是验证码类，一种是活动类。并且对应的，要将消息进行不同渠道的分发，还存在一个channelType，而他又存在QQ邮箱，手机短信、服务号三种不同的渠道。假如说我每增加一个渠道类型，我就必须再一次手动创建一个队列，那可太烦人了，并且还得新增一个新的监听器。而对应的，每个渠道商给的接口调用频率不同，如果我采用一个队列发送所有的渠道消息，这样一来，一旦某个服务被限流阻塞，那我核心服务就使用不了了。因此还需要对不同的渠道创建不同的队列进行消息隔离。这又面临一个问题，我怎么去动态的创建我自己的队列，不需要我每次都去手动新增队列、绑定队列？下面我说说自己的解决方案。当然，RabbitMQ的交换机的类型不同，对应的队列绑定方式也不太一样.

先复习复习，这些声明创建绑定方式

RabbitMQ中，声明队列以及绑定队列的几种方式

使用RabbitManager可视化图形界面创建、绑定队列

在queues中创建好队列

在exchanges中创建好交换机

在queues中，找到自己创建的队列，点进去，绑定上路由和交换机

或者

在exchanges中，找到自己创建的交换机，点进去，绑定上路由和交换机

反正无论你从哪进去，只要能将queue绑到交换机上，那就没问题。注意一些参数的选择。如果不理解可以看看我之前的文章。

写一个Config类，在Config类中声明队列绑定交换机

示例如下

/**  
* 将队列与交换机绑定  
*/  
@Configuration  
public class RabbitmqConfig {  
  
    //声明sms交换机  
    @Bean(RabbitmqConstant.SMS_EXCHANGE_NAME)  
    public Exchange SMS_EXCHANGE(){  
        //durable(true) 持久化，mq重启之后交换机还在  
        return ExchangeBuilder  
        .topicExchange(RabbitmqConstant.SMS_EXCHANGE_NAME)  
        .durable(true)  
        .build();  
    }  

    //声明sms队列  
    @Bean(RabbitmqConstant.SMS_QUEUE_NAME)  
    public Queue SMS_QUEUE(){  
        return new Queue(RabbitmqConstant.SMS_QUEUE_NAME);  
    }  


    //sms队列绑定sms交换机，指定routingKey  
    @Bean  
    public Binding BINDING_ROUTING_KEY_SMS(@Qualifier(RabbitmqConstant.SMS_QUEUE_NAME) Queue queue,  
    @Qualifier(RabbitmqConstant.SMS_EXCHANGE_NAME) Exchange exchange){  
        return BindingBuilder  
        .bind(queue)  
        .to(exchange)  
        .with(RabbitmqConstant.SMS_ROUTING_KET_FOR_REGISTER)  
        .noargs();  
    }  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

直接在@RabbitListener中声明并且绑定队列

@RabbitListener(bindings = @QueueBinding(  
value = @Queue(value = "${spring.rabbitmq.queues}", durable = "true"),  
exchange = @Exchange(value = "${rabbitmq.exchange.name}", type = ExchangeTypes.TOPIC),  
key = "${rabbitmq.routing.key}"  
))  
public void onMessage(Message message) {  
   //收到消息后处理...
}
1
2
3
4
5
6
7
8

RabbitAdmin的使用

RabbitAdmin 是 Spring AMQP 框架中的一个重要组件，用于简化和管理 RabbitMQ 的配置和操作。它提供了一种方便的方式来声明队列、交换机和绑定等 RabbitMQ 的对象，并使用这些对象进行消息的发送和接收。

RabbitAdmin 可以自动创建和绑定队列、交换机和绑定，这样可以在应用程序启动时自动配置 RabbitMQ 的相关对象，而不需要手动去创建和配置这些对象。

使用 RabbitAdmin，你可以通过编程的方式在应用程序中进行 RabbitMQ 的配置和操作，而不需要手动通过管理界面或命令行工具去创建和管理 RabbitMQ 的对象。

以下是 RabbitAdmin 的一些常见用法：

1. 声明队列：使用 declareQueue() 方法来声明队列，可以设置队列的名称、是否持久化、是否独占等属性。
2. 声明交换机：使用 declareExchange() 方法来声明交换机，可以设置交换机的类型、是否持久化等属性。
3. 声明绑定：使用 declareBinding() 方法来声明绑定，将交换机和队列通过指定的路由键绑定在一起。
4. 发送消息：使用 convertAndSend() 方法来向指定的交换机发送消息。
5. 接收消息：使用 receiveAndConvert() 方法来接收消息，并将消息转换为指定的 Java 对象。

一个问题

我的一个项目，对不同的业务的发送队列不一样。按照前文所说，我需要手写9个@RabbitListener，我真的会吐。下面看看我是怎么解决的。

Spring SpEL

Spring SpEL（Spring Expression Language，Spring 表达式语言）是 Spring 框架中提供的一种强大的表达式语言。它在功能上比 EL（Expression Language）更为强大，并且可以在更多的场景中使用。

Spring SpEL 可以在 Spring 应用程序的配置文件中使用，用于定义和处理字符串模板、计算表达式和访问对象属性等。 SpEL 提供了一套丰富的语法和功能，使得在 Spring 配置文件或 Spring Boot 的注解中可以进行更灵活和动态的编程。如果兴趣可以去看看这篇文章的详解使用。

深入掌握Spring SpEL及其实战应用

我这里就说一些我会用的
。SpEl可以使用在注解上，可以动态的写一个表达式，并且解析。表达式就类似于#{1 + 1 }、
#{beanName.get()}方法调用，或者是读取配置文件#{‘${rabbimq.name}’}。表达式可以调用SpringIOC容器
中的Bean的方法，下面我就用到了。

RabbitAdmin 配合 SpringEl表达式 和 @RabbitListener 以及Spring多例Bean，动态创建队列并且声明队列监听Bean

第一步，配合SpringBean的生命周期中，@PostConsturct注解，这个注解会在Bean属性赋值之后调用。因此，我拿他来用RabbitAdmin初始化我的队列
工具类，负责从数据库中获取队列的一些拼接段，并且将其拼接好，放在集合中。

@Component("queueNameMappingUtils")  
public class QueueNameMappingUtils {  
    /**  
    * 下标(用于迭代groupIds位置)  
    */  
    private static Integer index = 0;  

    private static List<String> queueNames = getAllQueueNames();  

    /**  
    * 获取所有队列名称
    */  
    public static List<String> getAllQueueNames(){  
        List<String> queueNames = new ArrayList<>();  
        ChannelType[] channelTypes = ChannelType.values();  
        MessageType[] messageTypes = MessageType.values();  
        //将他们组合起来  
        for (ChannelType channelType : channelTypes) {  
            for (MessageType messageType : messageTypes) {  
                groupIds.add(channelType.getCodeEn() + "." + messageType.getCodeEn());  
            }  
        }  
        return queueNames;  
    }  
    /**  
    * 一个一个获取  
    */  
    public String get(){  
        return queueNames.get(index++);  
    }  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

@Service  
@Slf4j  
public class RabbitMqReceiverInit{
    /**  
    * 上下文容器，用来获取实例  
    */  
    @Autowired  
    private ApplicationContext applicationContext;  

    /**  
    * 编程的方式绑定队列  
    */  
    @Autowired  
    private RabbitAdmin rabbitAdmin;  

    @Value("${message.push.rabbitmq.exchange.name}")  
    private String exchangeName;  
    @Value("${message.pusg.rabbitmq.queue.name.prefix}")  
    private String queuePrefix;  
    @Value("${message.push.rabbitmq.routing.key.prefix}")  
    private String routingKeyPrefix;  


    /**  
    * 获取得到所有的groupId  
    */  
    private static List<String> queueNames = QueueNameMappingUtils.getAllQueueNames();  

    /**  
    * 创建出对象，这个是在bean属性注入之后搞定的  
    */  
    @PostConstruct  
    public void init() {  
        for (int i = 0; i < queueNames.size(); i++) {  
            applicationContext.getBean(RabbitMqReceiver.class);  
        }  
        //申明交换机  
        rabbitAdmin.declareExchange(new TopicExchange(exchangeName, true, false));  
        //开始生产队列  
        for (String s : queueNames) {  
        //队列名称  
        String queueName = queuePrefix + "." + s;  
        //路由名称  
        String routingKey = routingKeyPrefix + "." + s;  
        //申明队列  
        rabbitAdmin.declareQueue(new Queue(queueName, true));  
        //绑定上交换机  
        rabbitAdmin.declareBinding(new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null));  
        }  
    }  

    @Bean  
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {  
        return new RabbitAdmin(connectionFactory);  
    }  
  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

第二，利用Bean的方法，以及下标偏移，不断获取下一个组装好的队列名称，并且通过SpringSpEl不断访问，最终成功创建多个监听器，并且监听的是不同的队列

@Component  
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)  
public class RabbitMqReceiver {  
    @RabbitListener(queues = "#{'${message.push.rabbitmq.queue.name.prefix}'+'.'+ queueNameMappingUtils.get()}")  
    public void onMessage(Message message) {  
        //处理消息...
    }  
}
1
2
3
4
5
6
7
8