通化 网站建设,电脑制作软件的工具,论坛网站文本抓取怎么做,福泉市建设局网站我们在消费RabbitMQ消息的过程中#xff0c;有时候可能会想先暂停消费一段时间#xff0c;然后过段时间再启动消费者#xff0c;这个需求怎么实现呢#xff1f;我们可以借助RabbitListenerEndpointRegistry这个类来实现#xff0c;它的全类名是org.springframework.amqp.r… 我们在消费RabbitMQ消息的过程中有时候可能会想先暂停消费一段时间然后过段时间再启动消费者这个需求怎么实现呢我们可以借助RabbitListenerEndpointRegistry这个类来实现它的全类名是org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry通过这个类可以实现全部队列消息的启动、停止消费也可以实现指定队列消息的启动、停止消费。具体的原因感兴趣的话可以参考一下我前面的这篇博客(17)不重启服务动态调整RabbitMQ消费者数量里面有相应的源码分析。
停止、启动全部队列消费 RabbitListenerEndpointRegistry类提供了start()方法和stop()方法可以看到底层都是通过调用getListenerContainers()获取到所有队列的消费监听容器列表然后遍历挨个调用对应的start()方法和stop()方法。 Overridepublic void start() {for (MessageListenerContainer listenerContainer : getListenerContainers()) {startIfNecessary(listenerContainer);}}Overridepublic void stop() {for (MessageListenerContainer listenerContainer : getListenerContainers()) {listenerContainer.stop();}} 我们只需要获取到RabbitListenerEndpointRegistry对象然后调用其start()方法和stop()方法即可实现启动/停止所有队列消费。 实现代码如下所示
Resource
RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;RequestMapping(value /startStopAllConsumer)
ApiOperation(value 启动/暂停全部队列消息消费)
public Response startStopAllConsumer(RequestParam(value consumeSwitch, required true) boolean consumeSwitch) {log.info(启动/暂停全部队列消息消费,consumeSwitch:{},consumeSwitch);if(consumeSwitch){rabbitListenerEndpointRegistry.start();}else {rabbitListenerEndpointRegistry.stop();}return Response.success();
} 传入开关参数为false会停止所有队列消费者消费调用后控制台看到如下日志
2023-09-04 19:43:11.480 0800 [TID: N/A] [http-nio-8080-exec-4] INFO c.b.t.m.p.w.PayCashierMockController:67 - 启动/暂停全部队列消息消费,consumeSwitch:false 2023-09-04 19:43:11.556 0800 [TID: N/A] [http-nio-8080-exec-4] INFO o.s.a.r.l.SimpleMessageListenerContainer:586 - Waiting for workers to finish. 2023-09-04 19:43:12.352 0800 [TID: N/A] [http-nio-8080-exec-4] INFO o.s.a.r.l.SimpleMessageListenerContainer:589 - Successfully waited for workers to finish. 可以看到消息监听容器关闭的日志然后再传入开关参数为true调用后会启动所有队列消息消费。
停止、启动指定队列消费 上面提到了RabbitListenerEndpointRegistry.getListenerContainers()可以获取到所有队列的消费监听容器列表我们可以使用MessageListenerContainer中获取消费的队列名进行判断以实现指定队列的停止、启动消费。 实现代码如下所示
Resource
RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;RequestMapping(value /startStopConsumer)
ApiOperation(value 启动/暂停指定队列消息消费)
public Response startStopConsumer(RequestParam(value queueName, required false) String queueName,RequestParam(value consumeSwitch, required true) boolean consumeSwitch) {log.info(启动/暂停指定队列消息消费,consumeSwitch:{},queueName:{},consumeSwitch,queueName);//获取所有消息监听容器CollectionMessageListenerContainer listenerContainers rabbitListenerEndpointRegistry.getListenerContainers();for (MessageListenerContainer container : listenerContainers) {SimpleMessageListenerContainer con (SimpleMessageListenerContainer) container;//消息监听容器要消费的队列名称集合ListString queueNamesList Arrays.asList(con.getQueueNames());//判断容器中的队列名称是否包含需要调整的队列名参数if (queueNamesList.contains(queueName)) {if(consumeSwitch){con.start();}else{con.stop();}}}return Response.success();
}
传入开关参数为false停止pay_work_notify队列消费者消费调用后控制台看到如下日志
2023-09-04 19:51:37.130 0800 [TID: N/A] [http-nio-8080-exec-1] INFO c.b.t.m.p.w.PayCashierMockController:80 - 启动/暂停指定队列消息消费,consumeSwitch:false,queueName:pay_work_notify 2023-09-04 19:51:37.200 0800 [TID: N/A] [http-nio-8080-exec-1] INFO o.s.a.r.l.SimpleMessageListenerContainer:586 - Waiting for workers to finish. 2023-09-04 19:51:37.903 0800 [TID: N/A] [http-nio-8080-exec-1] INFO o.s.a.r.l.SimpleMessageListenerContainer:589 - Successfully waited for workers to finish. 可以看到消息监听容器关闭的日志然后再传入开关参数为true调用后会启动pay_work_notify队列消息消费。
