zoukankan      html  css  js  c++  java
  • SpringBoot学习(六)—— springboot快速整合RabbitMQ

    Rabbit MQ消息队列

    @

    简介

    优点

    erlang开发,并发能力强。

    社区活跃,使用的人多,稳定性较强。

    延时低

    缺点

    erlang语言开发的,国内精通的不多,日后定制开发困难。

    Rabbit MQ工作模式

    1,"Hello World!"模式

    在这里插入图片描述

    简单模式是Rabbit MQ最简单入门的案例,其中一个生产者,一个消费者。无需声明交换机(其实是有个默认的交换机),声明完一个队列以后,生产者往这里发送,消费者则监听该队列,对消息做出响应。

    应用场景:如两个用户之间的聊天。

    2,Work queues 模式

    在这里插入图片描述

    工作队列模式在工人之间分配任务(竞争的消费者模式),就是有多个消费者的简单模式,多个消费者一起监听该队列,Rabbit MQ 会确保队列中的某一条消息只会被某一个消费者消费。

    应用场景:如多个用户抢某个订单(高并发下争抢同一个资源的记得考虑加锁)

    3,Publish/Subscribe 模式

    在这里插入图片描述

    发布/订阅模式一次向许多消费者发送消息,还是一个生产者多个消费者,但是该模式需要显示声明交换机了,将创建的多个队列与该交换机进行绑定,生产者将消息发给交换机,由交换机发给已经绑定好的队列,对应的某队列的消费者消费该队列中的信息。

    应用场景:群发某个邮件,推送某条广告

    4,Routing 模式

    在这里插入图片描述

    路由模式是有选择地接收消息,还是显示声明交换机,将队列与交换机绑定,由生产者将消息发给交换机,由交换机发给对应的队列,由对应的消费者消费信息。但是队列绑定交换机时需要指定一个routingkey ,这样的话,生产者再向生产者发消息的时候,会判断具体的 routingkey 值,并将符合对应 routingkey 值的消息发送给对应的队列。

    应用场景:封装若干类错误类型通知

    5,Topics 模式

    在这里插入图片描述

    主题模式根据模式(主题)接收消息,同路由模式一样,只不过不再是根据具体的 routingkey 值做判断,是根据 routingkey 值模糊匹配的。其中,星号代表多个单词 , 井号代表一个单词。

    更加详细的资料可访问官网 : RabbitMQ官网

    交换机模式

    fanout模式:不处理路由键,即没有 routingkey 值,很像子网广播,每台子网内的主机都获得了一份复制的消息。

    direct模式:处理路由键,需要路由键匹配才能转发,即完全匹配上 routingkey 值。

    topic模式:处理路由键,需要路由键模糊匹配才能转发,即模糊匹配上 routingkey 值。

    引入RabbitMQ队列

    RabbitMQ 依赖与 erlang 语言,点击 官网下载链接 下载 erlang。

    点击 官网下载链接 下载RabbitMQ。

    安装两者的时候,一路默认设置即可。

    启动 RabbitMQ 服务,cmd模式下输入,进入默认安装路径

    C:Program Files (x86)RabbitMQ Server
    abbitmq_server-3.6.10sbin
    

    输入命令

    rabbitmq-plugins enable rabbitmq_management
    

    服务启动成功,如图所示
    在这里插入图片描述

    点击 http://localhost:15672/ ,如下图所示

    在这里插入图片描述

    在这里插入图片描述

    pom.xml中新增

    <dependency>
    	<groupId>org.springframework.boot</groupId>
    	<artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    

    application.properties中加入

    # RabbitMQ
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    # 并发消费者的初始化值
    spring.rabbitmq.listener.concurrency=10
    # 并发消费者的最大值
    spring.rabbitmq.listener.max-concurrency=20
    # 每个消费者每次监听时可拉取处理的消息数量
    spring.rabbitmq.listener.prefetch=5
    

    代码实战

    简单模式和工作队列模式没有指定交换机,rabbitMQ的精妙之处不能体现出来,而路由模式和主体模式都有 routingkey 值,故挑其一,本文以发布/订阅模式和路由模式为例,其余的都可类比。不全写,是因为这是快速上手使用系列,最短的时间精力明白来龙去脉,才是该系列的宗旨,人类畏惧晦涩冗长的,喜欢简短直接的。

    新增代码后的目录如下所示

    在这里插入图片描述

    RabbitConfig.java

    package com.example.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    
    @Configuration
    public class RabbitConfig {
    
        //==============发布/订阅模式==============
    
        //队列A
        @Bean()
        public Queue queueA() {
            return new Queue("fanoutQueueA");
        }
    
        //队列B
        @Bean()
        public Queue queueB() {
            return new Queue("fanoutQueueB");
        }
    
        //发布订阅模式下的 fanout 交换机
        @Bean
        public FanoutExchange fanoutExchange() {
            return new FanoutExchange("fanoutExchange");
        }
    
        // 把 队列A 和 fanout交换机 绑定在一起
        @Bean
        public Binding bindingWithQueueA() {
            return BindingBuilder.bind(queueA()).to(fanoutExchange());
        }
    
        // 把 队列B 和 fanout交换机 绑定在一起
        @Bean
        public Binding bindingWithQueueB() {
            return BindingBuilder.bind(queueB()).to(fanoutExchange());
        }
    
    
        //==============路由模式==============
        //队列C
        @Bean()
        public Queue queueC() {
            return new Queue("directQueueC");
        }
    
        //队列D
        @Bean()
        public Queue queueD() {
            return new Queue("directQueueD");
        }
    
        //发布订阅模式下的 direct 交换机
        @Bean
        public DirectExchange directExchange() {
            return new DirectExchange("directExchange");
        }
    
        // 把 队列C 和 direct交换机 绑定在一起
        @Bean
        public Binding bindingWithQueueC() {
            return BindingBuilder.bind(queueC()).to(directExchange()).with("directRoutingkey");
        }
    
        // 把 队列B 和 direct交换机 绑定在一起
        @Bean
        public Binding bindingWithQueueD() {
            return BindingBuilder.bind(queueD()).to(directExchange()).with("directRoutingkey");
        }
    
    }
    

    RabbitController.java

    package com.example.controller;
    
    import com.example.service.IRabbitProducerService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    
    @Controller
    @RequestMapping("rabbit")
    public class RabbitController {
    
        @Autowired
        private IRabbitProducerService rabbitProducerService;
    
        @RequestMapping("/fanout")
        public void fanout() {
            for (int i = 0; i < 5; i++) {
                this.rabbitProducerService.producerFanout("发布订阅模式下的第" + i + "条信息");
            }
        }
    
        @RequestMapping("/direct")
        public void direct() {
            for (int i = 0; i < 5; i++) {
                this.rabbitProducerService.producerDirect("路由模式下的第" + i + "条信息");
            }
        }
    
    }
    

    IRabbitProducerService.java

    package com.example.service;
    
    public interface IRabbitProducerService {
    
        void producerFanout(String message);
    
        void producerDirect(String message);
    
    }
    

    RabbitProducerServiceIml.java

    package com.example.service;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    @Service("rabbitProducerService")
    public class RabbitProducerServiceIml implements IRabbitProducerService {
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        public void producerFanout(String message) {
            rabbitTemplate.convertAndSend("fanoutExchange", null, message);
        }
    
        public void producerDirect(String message) {
            rabbitTemplate.convertAndSend("directExchange", "directRoutingkey", message);
        }
    
    }
    

    IRabbitConsumerService.java

    package com.example.service;
    
    public interface IRabbitConsumerService {
    
        void consumerFanoutA(String message);
    
        void consumerFanoutB(String message);
    
        void consumerDirectC(String message);
    
        void consumerDirectD(String message);
    
    }
    

    RabbitConsumerServiceIml.java

    package com.example.service;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RabbitConsumerServiceIml implements IRabbitConsumerService {
    
        @RabbitListener(queues = "fanoutQueueA")
        public void consumerFanoutA(String message) {
            System.out.println("消费者收到队列A中的消息:" + message);
        }
    
        @RabbitListener(queues = "fanoutQueueB")
        public void consumerFanoutB(String message) {
            System.out.println("消费者收到队列B中的消息:" + message);
        }
    
        @RabbitListener(queues = "routingQueueC")
        public void consumerDirectC(String message) {
            System.out.println("消费者收到队列C中的消息:" + message);
        }
    
        @RabbitListener(queues = "routingQueueD")
        public void consumerDirectD(String message) {
            System.out.println("消费者收到队列D中的消息:" + message);
        }
    
    }
    

    启动项目后,浏览器输入 http://localhost:8080/rabbit/fanout ,(虽然前台会报错,因为这个请求只是触发消息的生产,并没有对应的界面展示。生产出来的消息经过交换机传给队列,消费者则监听队列,做出对应的响应。)发布/订阅模式下使用 fanout 交换机生产消费消息的效果图如下所示;

    在这里插入图片描述

    浏览器输入 http://localhost:8080/rabbit/direct ,发路由模式下使用 direct 交换机生产消费消息的效果图如下所示;

    在这里插入图片描述

    注:如果是根据本文系列文章来的,因为一开始就配置好了spring security,所以记得将该地址配给所登录的用户。或者开一个超级管理员账号,可以访问项目的任意目录,使用该管理员账号访问这些地址。

    到这里,篇幅已经很长了,本来想准备写完死信队列,和消息确认机制。想想这是快速整合上手系列,就把深入的内容放到踩坑深入系列,尽情期待后续开辟新的系列文章吧。

  • 相关阅读:
    HashCode的作用
    为什么重写equals一定要重写hashcode?
    java timer 定时器原理
    java内部类作用
    jeesite添加多数据源
    Eclipse之安装Jadclipse插件直接查看class文件
    NIO和IO的主要区别
    sql 语句
    HTTP Error 500.0
    基于.NetCore3.1系列 ——认证授权方案之Swagger加锁
  • 原文地址:https://www.cnblogs.com/lgx211/p/11958707.html
Copyright © 2011-2022 走看看