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

Spring中如何动态的创建、监听MQ以及创建Exchange

news 2025/7/17 1:30:06

在这里插入图片描述

文章目录

前言
动态创建和管理Exchange、Queue
动态消费Queue
结论

前言

前面我们学习 RabbitMQ 的时候，都是在编译的时候就确定了Exchange、Queue，也就是说我们需要在程序启动之前就创建好需要的Exchange和Queue，但是实际使用的时候，我们往往不能够实现知道所有的Exchange和Queue，而是在运行时才能够确定这些信息，那么在 Spring 中如何动态的管理这些呢？

动态创建和管理Exchange、Queue

在 Spring 中动态的创建和管理Exchange、Queue主要依赖于 RabbitAdmin 类，这个 RabbitAdmin Spring 已经帮我们管理了，只要我们导入了 RabbitMQ 的依赖就可以使用 @Autowired 或者 @Resource 将 RabbitAdmin 依赖注入就可以了：

@Autowired
private RabbitAdmin rabbitAdmin;

注入 RabbitAdmin 之后，我们就来看看如何创建 Exchange，跟之前学习的差不多，这里的创建 Exchange 也叫做声明交换机，也就是 delareExchange()方法：

在这里插入图片描述
方法需要的参数类型是 Exchange 类型，也就是前面我们学习的 fanout、direct、topic 三种类型：

在这里插入图片描述

所以在声明交换机之前，需要先创建出一个对应类型的交换机：

public void createExchange(String name) {FanoutExchange exchange = new FanoutExchange(name + ".exchange");
}

创建完成之后，将这个交换机以参数的形式传入到 declareExchange 方法中：

public void createExchange(String name) {FanoutExchange exchange = new FanoutExchange(name + ".exchange");rabbitAdmin.declareExchange(exchange);
}

上面是声明交换机的方式，那么声明队列的方式也是类似的：先创建队列然后再声明队列：

Queue queue = new Queue("message.queue");
rabbitAdmin.declareQueue(queue);

声明完成Exchange和Queue之后，接下来就是声明Exchange和Queue的Binding关系了，声明绑定关系同样如此，还是先创建Binding然后是声明Binding：

Binding binding = new Binding("message.queue", Binding.DestinationType.QUEUE,name+".exchange","message",null);
rabbitAdmin.declareBinding(binding);

注意：因为declare 方法的第一个和第三个参数的类型都是 String 类型，所以为了区分出哪个参数是队列的名字，哪个参数的是交换机的名字，就是用了第二个参数来指明。如果先绑定的是 Queue，也就是第一个参数是队列的名字，那么第二个参数就是 Binding.DestinationType.QUEUE，如果第一个参数是交换机的名字，那么第二个参数就是 Binding.DestinationType.EXCHANGE。

动态消费Queue

和创建和管理Exchange、Queue一样，如果我们不想简单的消费队列，而是经过一些逻辑之后才进行消费的话，该如何做呢，之前都是通过在 config 类中通过 @RabbitListener 注解来监听队列，那么我们如何显示的调用方法来监听队列呢？

Spring amqp 为我们提供了一个 MessageListener 接口，我们需要实现这个接口中的 onMessage() 方法：

在这里插入图片描述
当通过 MessageListener 接口实现了消费者的主要消费逻辑之后，我们就需要设置监听器，设置监听器依靠于 SimpleMessageListenerContainer 类中的 setMessageListener() 方法，然后具体监听哪个队列就通过 addQueueNames() 方法来指定需要消费的队列：

@Component
public class ThreadPoolUtil {@Resource(name = "taskExecutor")private Executor taskExecutor;@Autowiredprivate SimpleMessageListenerContainer simpleMessageListenerContainer;@Autowiredprivate RabbitMessageQueueListener rabbitMessageQueueListener;@PostConstructpublic void init() {taskExecutor.execute(() -> {simpleMessageListenerContainer.setMessageListener(rabbitMessageQueueListener);simpleMessageListenerContainer.addQueueNames("message.queue");});}
}