浅谈spring-boot-rabbitmq动态管理的方法

使用spring boot + rabbitmq的时候,在开发过程中,可能会想要临时停用/启用监听,或修改监听消费者数量。如果每次修改都重启比较浪费时间,所以研究了一下不停机就启用停用监听或修改一些配置

一. 关于rabbitmq监听的配置

  1. 配置属性类:RabbitProperties,包含rabbitmq的认证、监听、发送者以及其他的一些配置
  2. 自动配置类:RabbitAutoConfiguration,主要配置rabbitmq的连接工厂和发送者等,不包含监听的配置
  3. rabbitmq监听的配置是RabbitAnnotationDrivenConfiguration,是通过RabbitAutoConfiguration引入的
@Configuration
@ConditionalOnClass({ RabbitTemplate.class, Channel.class })
@EnableConfigurationProperties(RabbitProperties.class)
@Import(RabbitAnnotationDrivenConfiguration.class)
public class RabbitAutoConfiguration {
  ...
}

RabbitAnnotationDrivenConfiguration中主要就是监听工厂的配置、监听工厂,但是这里也只是创建bean,并没有真正的初始化

通过配置里的bean类名,分析一下,rabbitmq的监听肯定是由监听工厂创建的,所以找到监听工厂SimpleRabbitListenerContainerFactory

@Bean
@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
 SimpleRabbitListenerContainerFactoryConfigurer configurer,
 ConnectionFactory connectionFactory) {
  SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  configurer.configure(factory, connectionFactory);
  return factory;
}

既然自动配置里面没有初始化监听,那就应该是在其他地方调用的,进入监听工厂类中,发现有initializeContainer(SimpleMessageListenerContainer instance)方法,猜测初始化肯定与这个方法有关,所以查看有哪些地方调用,于是找到RabbitListenerEndpointRegistry.createListenerContainer(RabbitListenerEndpoint endpoint,RabbitListenerContainerFactory<?> factory)方法中有创建监听容器和初始化的代码

/**
 * Create and start a new {@link MessageListenerContainer} using the specified factory.
 * @param endpoint the endpoint to create a {@link MessageListenerContainer}.
 * @param factory the {@link RabbitListenerContainerFactory} to use.
 * @return the {@link MessageListenerContainer}.
 */
protected MessageListenerContainer createListenerContainer(RabbitListenerEndpoint endpoint,
 RabbitListenerContainerFactory<?> factory) {
  MessageListenerContainer listenerContainer = factory.createListenerContainer(endpoint);
  if (listenerContainer instanceof InitializingBean) {
   try {
      ((InitializingBean) listenerContainer).afterPropertiesSet();
   }
   catch (Exception ex) {
      throw new BeanInitializationException("Failed to initialize message listener container", ex);
   }
  }
  int containerPhase = listenerContainer.getPhase();
  if (containerPhase < Integer.MAX_VALUE) { // a custom phase value
   if (this.phase < Integer.MAX_VALUE && this.phase != containerPhase) {
      throw new IllegalStateException("Encountered phase mismatch between container factory definitions: " +
       this.phase + " vs " + containerPhase);
   }
   this.phase = listenerContainer.getPhase();
  }
  return listenerContainer;
}

继续找调用这个方法的地方,找到RabbitListenerEndpointRegistrar.afterPropertiesSet()方法之后,发现调用的地方很多了

看看afterPropertiesSet方法,是InitializingBean接口中的,猜测应该是spring容器创建bean之后都会调用的bean初始化的方法,所以查找找到RabbitListenerEndpointRegistrar是在哪里创建的实例。原来是在RabbitListenerAnnotationBeanPostProcessor中的私有属性,而RabbitListenerAnnotationBeanPostProcessor是在RabbitBootstrapConfiguration这个自动配置里面初始化的,所以这就找到rabbitmq初始化监听的源头了

二. 动态管理rabbitmq监听

回到最初的问题,想要动态的启用停用mq的监听,所以先看看初始化配置的类,既然有初始化,那可能会有相关的管理,于是在RabbitListenerEndpointRegistry中找到了start()和stop()方法,里面有对监听容器进行操作,主要源码如下

/**
 * @return the managed {@link MessageListenerContainer} instance(s).
 */
public Collection<MessageListenerContainer> getListenerContainers() {
  return Collections.unmodifiableCollection(this.listenerContainers.values());
}

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

/**
 * Start the specified {@link MessageListenerContainer} if it should be started
 * on startup or when start is called explicitly after startup.
 * @see MessageListenerContainer#isAutoStartup()
 */
private void startIfNecessary(MessageListenerContainer listenerContainer) {
  if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
   listenerContainer.start();
  }
}

@Override
public void stop() {
  for (MessageListenerContainer listenerContainer : getListenerContainers()) {
   listenerContainer.stop();
  }
}

写个controller,注入RabbitListenerEndpointRegistry,使用start()和stop()对监听进行启用停用的操作,并且RabbitListenerEndpointRegistry实例还可以获取监听容器,对监听的一些参数也能进行修改,比如消费者数量。代码如下:

import java.util.Set;
import javax.annotation.Resource;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.itopener.framework.ResultMap;
/**
 * Created by fuwei.deng on 2017年7月24日.
 */
@RestController
@RequestMapping("rabbitmq/listener")
public class RabbitMQController {

  @Resource
  private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;

  @RequestMapping("stop")
  public ResultMap stop(){
   rabbitListenerEndpointRegistry.stop();
   return ResultMap.buildSuccess();
  }

  @RequestMapping("start")
  public ResultMap start(){
   rabbitListenerEndpointRegistry.start();
   return ResultMap.buildSuccess();
  }

  @RequestMapping("setup")
  public ResultMap setup(int consumer, int maxConsumer){
   Set<String> containerIds = rabbitListenerEndpointRegistry.getListenerContainerIds();
   SimpleMessageListenerContainer container = null;
   for(String id : containerIds){
   container = (SimpleMessageListenerContainer) rabbitListenerEndpointRegistry.getListenerContainer(id);
   if(container != null){
    container.setConcurrentConsumers(consumer);
    container.setMaxConcurrentConsumers(maxConsumer);
   }
   }
   return ResultMap.buildSuccess();
  }
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

您可能感兴趣的文章:

  • springboot集成rabbitMQ之对象传输的方法
  • springboot整合rabbitmq的示例代码
  • Spring Boot与RabbitMQ结合实现延迟队列的示例
  • spring boot集成rabbitmq的实例教程
  • 详解Spring Boot 配置多个RabbitMQ
  • spring boot整合RabbitMQ实例详解(Fanout模式)
  • Spring Boot整合RabbitMQ实例(Topic模式)
  • spring boot整合RabbitMQ(Direct模式)
  • 详解spring boot集成RabbitMQ
(0)

相关推荐

  • spring boot整合RabbitMQ实例详解(Fanout模式)

    1.Fanout Exchange介绍 Fanout Exchange 消息广播的模式,不管路由键或者是路由模式,会把消息发给绑定给它的全部队列,如果配置了routing_key会被忽略. 如上图所示,即当使用fanout交换器时,他会将消息广播到与该交换器绑定的所有队列上,这有利于你对单条消息做不同的反应. 例如存在以下场景:一个web服务要在用户完善信息时,获得积分奖励,这样你就可以创建两个对列,一个用来处理用户信息的请求,另一个对列获取这条消息是来完成积分奖励的任务. 2.代码示例 1).

  • springboot整合rabbitmq的示例代码

    概述 RabbitMQ是一个开源的消息代理和队列服务器,用来通过普通协议在完全不同的应用之间共享数据,或者简单地将作业队列以便让分布式服务器进行处理. 它现实了AMQP协议,并且遵循Mozilla Public License开源协议,它支持多种语言,可以方便的和spring集成. 消息队列使用消息将应用程序连接起来,这些消息通过像RabbitMQ这样的消息代理服务器在应用程序之间路由. 基本概念 Broker 用来处理数据的消息队列服务器实体 vhost 由RabbitMQ服务器创建的虚拟消息

  • Spring Boot整合RabbitMQ实例(Topic模式)

    1.Topic交换器介绍 Topic Exchange 转发消息主要是根据通配符. 在这种交换机下,队列和交换机的绑定会定义一种路由模式,那么,通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息. 在这种交换机模式下: 路由键必须是一串字符,用句号(.) 隔开,比如说 agreements.us,或者 agreements.eu.stockholm 等. 路由模式必须包含一个 星号(*),主要用于匹配路由键指定位置的一个单词,比如说,一个路由模式是这样子:agreements..b.*

  • 详解Spring Boot 配置多个RabbitMQ

    闲话 好久没有写博客了,6月份毕业,因为工作原因,公司上网受限,一直没能把学到的知识点写下来,工作了半年,其实学到的东西也不少,但是现在回忆起来的东西少之又少,有时甚至能在同个问题中踩了几次,越来越觉得及时记录一下学到的东西很重要. 好了,闲话少说,写下这段时间学习的东西,先记录一下用spring Boot配置多个RabbitMQ的情况... 最近公司新启动一个新平台的项目,需要用微服务这个这几年很火的概念来做,所以就学习了Spring Boot方面的知识,给同事展示Spring Boot的一些

  • spring boot整合RabbitMQ(Direct模式)

    springboot集成RabbitMQ非常简单,如果只是简单的使用配置非常少,springboot提供了spring-boot-starter-amqp项目对消息各种支持. 1.新建一个Spring Boot工程,命名为:"rabbitmq-hello". 在pom.xml中引入如下依赖内容,其中spring-boot-starter-amqp用于支持RabbitMQ. <dependency> <groupId>org.springframework.boo

  • Spring Boot与RabbitMQ结合实现延迟队列的示例

    背景 何为延迟队列? 顾名思义,延迟队列就是进入该队列的消息会被延迟消费的队列.而一般的队列,消息一旦入队了之后就会被消费者马上消费. 场景一:在订单系统中,一个用户下单之后通常有30分钟的时间进行支付,如果30分钟之内没有支付成功,那么这个订单将进行一场处理.这是就可以使用延时队列将订单信息发送到延时队列. 场景二:用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作.这时候就可以将用户指令发送到延时队列,当指令设定的时间到了再将指令推送到只能设备. 延迟队列能做什么? 延迟队列多用于需

  • 详解spring boot集成RabbitMQ

    RabbitMQ作为AMQP的代表性产品,在项目中大量使用.结合现在主流的spring boot,极大简化了开发过程中所涉及到的消息通信问题. 首先正确的安装RabbitMQ及运行正常. RabbitMQ需啊erlang环境,所以首先安装对应版本的erlang,可在RabbitMQ官网下载 # rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm 使用yum安装RabbitMQ,避免缺少依赖包引起的安装失败 # yum install rabbitmq-s

  • springboot集成rabbitMQ之对象传输的方法

    rabbitMQ的安装方法网上有很多教程,这里就不重复了. 在springboot上使用rabbitMQ传输字符串和对象,本文所给出的例子是在两个不同的项目之间进行对象和和字符串的传输. rabbitMQ的依赖(在两个项目中一样的配置): <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId>

  • spring boot集成rabbitmq的实例教程

    一.RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache). 消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下: 从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息

  • 浅谈spring boot 1.5.4 异常控制

    spring boot 已经做了统一的异常处理,下面看看如何自定义处理异常 1.错误码页面映射 1.1静态页面 必须配置在 resources/static/error文件夹下,以错误码命名 下面是404错误页面内容,当访问一个不存在的链接的时候,定位到此页 <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Not F

  • Quartz+Spring Boot实现动态管理定时任务

    项目实践过程中碰到一个动态管理定时任务的需求:针对每个人员进行信息的定时更新,具体更新时间可随时调整.启动.暂定等. 思路 将每个人员信息的定时配置保存到数据库中,这样实现了任务的动态展示和管理.任务的每一次新增或变更,都会去数据库变更信息. 设置一个统一的任务管理器,专门负责动态任务的增删改查. POM依赖 <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://mav

  • 浅谈spring boot 集成 log4j 解决与logback冲突的问题

    现在很流行springboot的开发,小编闲来无事也学了学,开发过程中遇见了log4j日志的一个小小问题,特此记载. 首先在pox.xml中引入对应的maven依赖: <!-- 引入log4j--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency&g

  • 浅谈Spring Boot: 接口压测及简要优化策略

    工程做好之后,需要对接口进行压力测试.可以自己编写线程池模拟多用户访问测试,也可以使用jmeter进行压测.jmeter的好处是测试方便,并且有完善的结果分析功能.本次采用jmeter进行压力测试. 1.准备数据,为了测试准备200w条以上的数据.一个简单的方法是使用下面的sql快速创建. INSERT INTO table (user_name,address) SELECT user_name, address FROM table; 但这样创建的数据不同记录的重复部分太多,和实际业务不太相

  • 浅谈Spring Boot日志框架实践

    Java应用中,日志一般分为以下5个级别: ERROR 错误信息 WARN 警告信息 INFO 一般信息 DEBUG 调试信息 TRACE 跟踪信息 Spring Boot使用Apache的Commons Logging作为内部的日志框架,其仅仅是一个日志接口,在实际应用中需要为该接口来指定相应的日志实现. SpringBt默认的日志实现是Java Util Logging,是JDK自带的日志包,此外SpringBt当然也支持Log4J.Logback这类很流行的日志实现. 统一将上面这些 日志

  • 浅谈Spring Boot中如何干掉if else的方法

    前言 看到crossoverJie的文章<利用策略模式优化过多 if else 代码>后受到启发,可以利用策略模式简化过多的if else代码,文章中提到可以通过扫描实现处理器的自注册,我在这里介绍在Spring Boot框架中的实现方法. 需求 这里虚拟一个业务需求,让大家容易理解.假设有一个订单系统,里面的一个功能是根据订单的不同类型作出不同的处理. 订单实体: public class OrderDTO { private String code; private BigDecimal

  • 浅谈Spring Boot 整合ActiveMQ的过程

    RabbitMQ是比较常用的AMQP实现,这篇文章是一个简单的Spring boot整合RabbitMQ的教程. 安装ActiveMQ服务器,(也可以不安装,如果不安装,会使用内存mq) 构建Spring boot项目,增加依赖项,只需要添加这一项即可 <!-- 添加acitivemq依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring

  • 浅谈Spring Boot 2.0迁移指南主要注意点

    Spring官方的Spring Boot 2变动指南,主要是帮助您将应用程序迁移到Spring Boot 2.0,变化部分还是很多很细节的,摘录主要点如下: Spring Boot 2.0需要Java 8或更高版本.不再支持Java 6和7.它还需要Spring Framework 5.0,许多配置属性被重新命名/删除,开发者需要更新他们的application.properties/ application.yml相应.为了帮助您,Spring Boot提供了一个新spring-boot-pr

  • 浅谈Spring boot cache使用和原理

    缓存要解决的问题:一个程序的瓶颈在于数据库,我们也知道内存的速度是大大快于硬盘的速度的.当我们需要重复地获取相同的数据的时候,我们一次又一次的请求数据库或者远程服务,导致大量的时间耗费在数据库查询或者远程方法调用上,导致程序性能的恶化,这便是数据缓存要解决的问题. 类似的缓存技术有:Redis.EhCache.Guava等,现在一般常用的为Redis. Spring 3.1 引入了激动人心的基于注释(annotation)的缓存(cache)技术,它本质上不是一个具体的缓存实现方案(例如EHCa

  • 浅谈Spring Boot 微服务项目的推荐部署方式

    如果开发过spring boot的程序,应该都知道,使用spring boot官方的maven打包插件(spring-boot-maven-plugin) 来打包,打出来的jar包一般有40M以上. 如果公司的服务器上传带宽不高,那么手动上传一个jar或者jenkins部署一次jar,都是非常痛苦的........ 但是,如果打包的时候不引入lib,那么打出来的jar包一般只有几十k而已,非常小,想怎么传就怎么传......... 本文会提供一个bash启动脚本,只需要稍做更改,即可适应你的程序

随机推荐