美文网首页
Springboot集成RabbitMQ

Springboot集成RabbitMQ

作者: squirrels | 来源:发表于2020-03-21 23:39 被阅读0次

    一.首先我们简单了解一下消息中间件的应用场景

    1) 异步处理

    场景说明:用户注册后,需要发注册邮件和注册短信,传统的做法有两种1.串行的方式;2.并行的方式
    (1)串行方式:将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务全部完成后才返回给客户端。 这有一个问题是,邮件,短信并不是必须的,它只是一个通知,而这种做法让客户端等待没有必要等待的东西.


    串行方式.png

    (2)并行方式:将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提高处理的时间。


    并行方式.png

    假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms。虽然并性已经提高的处理时间,但是,前面说过,邮件和短信对我正常的使用网站没有任何影响,客户端没有必要等着其发送完成才显示注册成功,应该是写入数据库后就返回.
    (3)消息队列
    引入消息队列后,把发送邮件,短信不是必须的业务逻辑异步处理

    引入消息队列.png

    由此可以看出,引入消息队列后,用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计),引入消息队列后处理后,响应时间是串行的3倍,是并行的2倍。

    2) 应用解耦

    场景:双11是购物狂节,用户下单后,订单系统需要通知库存系统,传统的做法就是订单系统调用库存系统的接口.


    传统方式.png

    这种做法有一个缺点:

    • 当库存系统出现故障时,订单就会失败。
    • 订单系统和库存系统高耦合.
      引入消息队列


      引入消息队列.png

    订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。
    库存系统:订阅下单的消息,获取下单消息,进行库操作。
    就算库存系统出现故障,消息队列也能保证消息的可靠投递,不会导致消息丢失。

    3) 流量削峰

    流量削峰一般在秒杀活动中应用广泛
    场景:秒杀活动,一般会因为流量过大,导致应用挂掉,为了解决这个问题,一般在应用前端加入消息队列。 作用:

    • 可以控制活动人数,超过此一定阀值的订单直接丢弃(我为什么秒杀一次都没有成功过呢)
    • 可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单)
    示意图.png

    1.用户的请求,服务器收到之后,首先写入消息队列,加入消息队列长度超过最大值,则直接抛弃用户请求或跳转到错误页面.
    2.秒杀业务根据消息队列中的请求信息,再做后续处理.

    二.各种消息中间件性能的比较:

    • TPS比较 :ZeroMq 最好,RabbitMq 次之, ActiveMq 最差。

    • 持久化消息比较:zeroMq不支持,activeMq和rabbitMq都支持。持久化消息主要是指:MQ down或者MQ所在的服务器down了,消息不会丢失的机制。

    • 可靠性、灵活的路由、集群、事务、高可用的队列、消息排序、问题追踪、可视化管理工具、插件系统、社区:RabbitMq最好,ActiveMq次之,ZeroMq最差。

    • 高并发:从实现语言来看,RabbitMQ最高,原因是它的实现语言是天生具备高并发高可用的erlang语言。

    综上所述:RabbitMQ的性能相对来说更好更全面,是消息中间件的首选。

    三.接下来我们在springboot当中整合使用RabbitMQ

    本次实例教程需要创建2个springboot项目,一个 rabbitmq-provider (生产者),一个rabbitmq-consumer(消费者)。

    第一步:导入maven依赖

    <dependency>    
      <groupId>org.springframework.boot</groupId>   
      <artifactId>spring-boot-starter-amqp</artifactId> 
      <version>1.5.2.RELEASE</version>
    </dependency>
    <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-web</artifactId>
     </dependency>
    

    第二步:在application.properties文件当中引入RabbitMQ基本的配置信息

    #对于rabbitMQ的支持
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=admin
    

    第三步:编写RabbitConfig类,类里面设置很多个EXCHANGE,QUEUE,ROUTINGKEY,是为了接下来的不同使用场景。把交换机,队列,通过路由关键字进行绑定

    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.beans.factory.config.ConfigurableBeanFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Scope;
    /**
     * Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线,保证数据能按照指定的方式进行传输, 
     * Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列。 
     * Queue:消息的载体,每个消息都会被投到一个或多个队列。 
     * Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来. 
     * Routing Key:路由关键字,e
     * xchange根据这个关键字进行消息投递。 
     * vhost:虚拟主机,一个broker里可以有多个vhost,用作不同用户的权限分离。 
     * Producer:消息生产者,就是投递消息的程序. 
     * Consumer:消息消费者,就是接受消息的程序. 
     * Channel:消息通道,在客户端的每个连接里,可建立多个channel.
     * */
    @Configuration
    public class RabbitConfig {
        private  final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        @Value("${spring.rabbitmq.host}")
        private String host;
    
        @Value("${spring.rabbitmq.port}")
        private int port;
    
        @Value("${spring.rabbitmq.username}")
        private String username;
    
        @Value("${spring.rabbitmq.password}")
        private String password;
    
        public static final String EXCHANGE_A = "my-mq-exchange_A";
        public static final String EXCHANGE_B = "my-mq-exchange_B";
        public static final String EXCHANGE_C = "my-mq-exchange_C";
    
        public static final String QUEUE_A = "QUEUE_A";
        public static final String QUEUE_B = "QUEUE_B";
        public static final String QUEUE_C = "QUEUE_C";
    
        public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";
        public static final String ROUTINGKEY_B = "spring-boot-routingKey_B";
        public static final String ROUTINGKEY_C = "spring-boot-routingKey_C";
    
    
        @Bean
        public CachingConnectionFactory connectionFactory(){
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost("/");
            connectionFactory.setPublisherConfirms(true);
            return connectionFactory;
        }
    
        @Bean
        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
        public RabbitTemplate rabbitTemplate(){
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
            return rabbitTemplate;
        }
    
        /**
         * 针对消费者配置
         * 1. 设置交换机类型
         * 2. 将队列绑定到交换机
         * FanoutExchange: 将消息分发到所有的绑定队列,无routingkey的概念
         * HeadersExchange :通过添加属性key-value匹配
         * DirectExchange:按照routingkey分发到指定队列
         * TopicExchange:多关键字匹配
         * @return
         */
        @Bean
        public DirectExchange defaultExchange(){
            return new DirectExchange(EXCHANGE_A);
        }
    
        /**
         * 获取队列A
         * @return
         */
        @Bean
        public Queue queueA(){
            return new Queue(QUEUE_A,true);//队列持久
        }
    
        @Bean
        public Queue queueB(){
            return new Queue(QUEUE_B,true);//队列持久
        }
    
        @Bean
        public Binding binding(){
            return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);
    
        }
    
        @Bean
        public Binding bindingB(){
            return BindingBuilder.bind(queueB()).to(defaultExchange()).with(RabbitConfig.EXCHANGE_B);
        }
    }
    

    第四步:编写消息的生产者

    import com.zss.study.mq.mymqtest.myconfig.RabbitConfig;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    import java.util.UUID;
    
    @RestController
    public class MsgProducer implements RabbitTemplate.ConfirmCallback {
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        //由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE,所以不能自动注入
        private RabbitTemplate rabbitTemplate;
        /**
         * 构造方法注入rabbitTemplate
         * @param rabbitTemplate
         */
        @Autowired
        public MsgProducer(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            rabbitTemplate.setConfirmCallback(this);
        }
        @GetMapping("/sendMsg")
        public String   sendMsg(){
            CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
            //把消息放入ROUTINGKEY_A对应的队列当中去,对应的是队列A
            Object content = "sssssss";
            rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A,RabbitConfig.ROUTINGKEY_A,content,correlationId);
            return "ok";
        }
        /**
         * 回调
         * @param correlationData
         * @param ack
         * @param cause
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            logger.info("回调id:"+correlationData);
            if(ack){
                logger.info("消息成功消费");
            }else {
                logger.info("消息消费失败"+cause);
            }
        }
    }
    

    相关文章

      网友评论

          本文标题:Springboot集成RabbitMQ

          本文链接:https://www.haomeiwen.com/subject/kdqpyhtx.html