(18)不重启服务动态停止、启动RabbitMQ消费者-Toy模板网

这篇具有很好参考价值的文章主要介绍了(18)不重启服务动态停止、启动RabbitMQ消费者。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

我们在消费RabbitMQ消息的过程中，有时候可能会想先暂停消费一段时间，然后过段时间再启动消费者，这个需求怎么实现呢？我们可以借助RabbitListenerEndpointRegistry这个类来实现，它的全类名是org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry，通过这个类可以实现全部队列消息的启动、停止消费，也可以实现指定队列消息的启动、停止消费。具体的原因感兴趣的话可以参考一下我前面的这篇博客(17)不重启服务动态调整RabbitMQ消费者数量，里面有相应的源码分析。

停止、启动全部队列消费

RabbitListenerEndpointRegistry类提供了start()方法和stop()方法，可以看到底层都是通过调用getListenerContainers()获取到所有队列的消费监听容器列表，然后遍历挨个调用对应的start()方法和stop()方法。

	@Override
	public void start() {
		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
			startIfNecessary(listenerContainer);
		}
	}

	@Override
	public 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);
	//获取所有消息监听容器
	Collection<MessageListenerContainer> listenerContainers = rabbitListenerEndpointRegistry.getListenerContainers();
	for (MessageListenerContainer container : listenerContainers) {
		SimpleMessageListenerContainer con = (SimpleMessageListenerContainer) container;
		//消息监听容器要消费的队列名称集合
		List<String> 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队列消息消费。文章来源地址https://www.toymoban.com/news/detail-707658.html

到了这里，关于(18)不重启服务动态停止、启动RabbitMQ消费者的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！