微服务08-认识和使用SpringAMQP

2023-09-13 17:12:44

1.AMQP的认识

1.1 介绍

AMQP是什么？看完你就知道了_hello_读书就是赚钱的博客-CSDN博客_amqp

在这里插入图片描述
好处：
什么connection：消息队列的连接、channel：服务发送接收消息的通道、Queue：消息队列——>这些你都不需要自己编写

工作过程：
发布者（Publisher）发布消息（Message），经由交换机（Exchange）。

交换机根据路由规则将收到的消息分发给与该交换机绑定的队列（Queue）。

最后 AMQP 代理会将消息投递给订阅了此队列的消费者，或者消费者按照需求自行获取。

深入理解：
1、发布者、交换机、队列、消费者都可以有多个。同时因为 AMQP 是一个网络协议，所以这个过程中的发布者，消费者，消息代理 可以分别存在于不同的设备上。

2、发布者发布消息时可以给消息指定各种消息属性（Message Meta-data）。有些属性有可能会被消息代理（Brokers）使用，然而其他的属性则是完全不透明的，它们只能被接收消息的应用所使用。

3、从安全角度考虑，网络是不可靠的，又或是消费者在处理消息的过程中意外挂掉，这样没有处理成功的消息就会丢失。基于此原因，AMQP 模块包含了一个消息确认（Message Acknowledgements）机制：当一个消息从队列中投递给消费者后，不会立即从队列中删除，直到它收到来自消费者的确认回执（Acknowledgement）后，才完全从队列中删除。

4、在某些情况下，例如当一个消息无法被成功路由时（无法从交换机分发到队列），消息或许会被返回给发布者并被丢弃。或者，如果消息代理执行了延期操作，消息会被放入一个所谓的死信队列中。此时，消息发布者可以选择某些参数来处理这些特殊情况。

1.2 实战：消息发送与消息接收

消息发送
在这里插入图片描述

@RunWith(SpringRunner.class)
@EnableRabbit
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    public void testSendMessageSimpleQueue(){
        String queueName="simple.queue";
        String messgae="Hello,SpringAMQP";
        rabbitTemplate.convertAndSend(queueName,messgae);
    }
}

可以发现队列simple.queue中 多了一条消息
在这里插入图片描述

AMOP如何发送消息？

引入amqp的starter依赖，然后我们配置RabbitMQ的地址，最后利用RabbitTemplate中的convertAndSend方法发送消息

在这里插入图片描述

消息接收：
在这里插入图片描述

1、首先进行yaml配置，RabbitMQ连接信息

2、新建一个组件，里面编写消费逻辑（利用@RabbitListener监听队列，只要队列一有消息就进行接收）

 
@Component
public class SpringRabbitListener {
 
    /**
     * @RabbitListener监听队列
     * @param msg
     */
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg){
        System.out.println("消费者接收到的simple.queue中消息为："+msg);
    }
}

在这里插入图片描述
在这里插入图片描述

注意：消息一旦被消费就会从队列删除，RabbitMQ没有消息回溯功能

2.WorkQueue

场景：
采用多个工作队列处理消息，避免消息堆积；

在这里插入图片描述

2.1 实战：多个消费者绑定一个队列

在这里插入图片描述

1.首先配置类中把队列支棱起来，Publisher发布消息给消息队列simple.queue

@RunWith(SpringRunner.class)
@EnableRabbit
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    public void testSendMessageWorkQueue() throws InterruptedException {
        String queueName="simple.queue";
        String messgae="华为手机，遥遥领先--";
        for (int i = 0; i < 50; i++) {
            rabbitTemplate.convertAndSend(queueName,messgae+i);
            Thread.sleep(20);
        }
    }

2.consumer服务中定义两个消费者，利用@RabbitListener监听队列

@Component
public class SpringRabbitListener {
    /**
     * 我们的思想是：两个消费者，一个1s能够消费50条，一个消费1s5条，一共50条信息，能者多劳
     * 现实：默认是平均分配->造成处理消息时间过长（因为第二个消费者处理消息很慢）
     * @param msg
     * @throws InterruptedException
     */
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue(String msg) throws InterruptedException {
        System.out.println("消费者1......接收到消息为："+msg+"当前时间："+ LocalTime.now());
        //每s50个消息
        Thread.sleep(20);
    }
 
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2.......接收到消息为："+msg+"当前时间："+ LocalTime.now());
        Thread.sleep(200);
    }

在这里插入图片描述
我们可以采用prefetch设置消费者预取的消息数量，这样就不会出现平均分配消息的现象了——>而是根据每个消费者的能力来处理消息

在这里插入图片描述

3.Fanout Exchange

前言：

发布（Publish）和订阅（Subscribe）

因为一个消息只可能被一个消费者消费，消费完就会删除该消费，所以我们需要利用发布和订阅来进行处理，这样就可以将同一个消息发送给多个消费者——>
利用exchange交换机将消息发送给多个队列，然后队列是可以将消息进行储存的，发给多个消费者

在这里插入图片描述
常见的exchange类型：Fanout（广播）、Direct（路由）、Topic（话题）

注意：exchange负责消息路由，而不是储存，路由失败则会造成消息丢失（也就是说路由到哪个消息队列）。

3.1 介绍

exchange: 只能作为消息的转发，记住不能作为消息的缓存，如果路由失败消息就丢了

交换机exchange：通过FanoutExchange返回交换机

消息队列Queue：通过Queue返回消息队列

将消息队列与交换机进行绑定Bingding：可以通过Binding中的bind方法进行绑定

3.2 实战

在这里插入图片描述

1:
在这里插入图片描述
consumer服务中声明交换机和队列，将队列与交换机进行绑定（利用Binding中bind方法）

@Configuration
public class FanoutConfig {
    //itcast.fanout交换机
    @Bean
    public FanoutExchange fanoutExchange(){
       return new FanoutExchange("itcast.fanout");
    }
 
    //fanout.queue1任务队列
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
 
    //消息队列2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
 
    //将消息队列与交换机进行绑定:类型和名称要保持一致，不然无法注入
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }
 
    //第二个消息队列与交换机进行绑定
    @Bean
    public Binding fanouting2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }
}

2.在这里插入图片描述

@Component
public class SpringRabbitListener {
    /**
     * 以下是利用交换机通知信息给处理消息的消费者
     */
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg){
        System.out.println("消费者接收到的fanout.queue1的消息为："+msg);
    }
 
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg){
        System.out.println("消费者收到fanout.queue2的消息为："+msg);
    }
}

3.在这里插入图片描述

@RunWith(SpringRunner.class)
@EnableRabbit
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    /**
     * 发送信息给交换机exchange
     */
    @Test
    public void testSendFanoutExchange(){
        //1.交换机名称
        String exchangeName="itcast.fanout";
        //2.消息
        String message="hello,every one!";
        //3.发送消息
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }
}

在这里插入图片描述

3.3 总结

在这里插入图片描述

4.Direct Exchang

DirectExchang：可以根据规则路由到指定的消息队列

4.1实战

在这里插入图片描述
1.在consumer服务中声明Exchange交换机与不同key之消息队列的绑定，利用@RabbitListener->还监听了一波消息队列

对比与之前的Fanout，它们的队列以及交换机是在一个配置类中定义并绑定的，利用了Binding，并注入容器

消费者代码
在这里插入图片描述

提供者代码

@Test
    public void testSendDirectExchange() {
        //交换机名称
        String exchangeName = "denghao.direct";
        //消息
        String message = "hello, blue!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"blue",message);

    }

个人编写与老师资料略有不同 敬请谅解
在这里插入图片描述

Direct交换机与Fanout交换机之间的差异+@RabbitListener注解中声明队列与交换机的常见注解
在这里插入图片描述

5.TopicExchange在这里插入图片描述

为什么要用Topic交换机，因为以后有的时候，比如地区天气新闻，有“长沙的，成都的，四川的，重庆的。。。。。。 如果使用DirectExchange的话，那么会要定义很多很多的key 而用TopicExchange的话 可以直接用 #.news 表示接收所有地方的天气新闻”

5.1 实战

介绍：

两个队列：一个是中国的新闻，一个所有的新闻消息；

两个消费者分别监听这两个队列

Publisher将消息发送给交换机Exchange，然后交换机根据key（+通配符）->决定将消息路由到哪个队列中

区别：

与DirectExchange区别就是，DirectExchange没有通配符，TopicExchange有通配符：xxx.xxx

最明显的地方就是：队列与交换机进行绑定时，key不一样

在这里插入图片描述

消费者的监听器

  @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "denghao.topic",type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息：【"+ msg +"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "denghao.topic",type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【"+ msg +"】");
    }

服务者测试方，发布消息：

@Test
    public void testSendTopicExchange() {
        //交换机名称
        String exchangeName = "denghao.topic";
        //消息
        String message1 = "红色天气预报警告，马上长沙迎来大升温，明天提升六度，请注意，不要感冒，不要鼻炎!!!";
        String message2 = "重磅新闻来袭!!!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"china.news",message1);
        rabbitTemplate.convertAndSend(exchangeName,"jp.news",message2);

    }

结果
在这里插入图片描述
在这里插入图片描述

6. 消息转换器

场景： 因为我们Publisher服务者发送消息到消息队列（将消息序列化为二进制），消息是乱码的——>因为消息类型content-type是java序列化变来的类型：缺点：消息体大、传输速度慢、安全问题、占内存；

在这里插入图片描述
处理：

1.我们定义一个bean，利用boot的自动配置思想，将默认的替换

@Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
      return new Jackson2JsonMessageConverter();
    }

2.然后导入依赖，Jacksonxxxx

  <!--序列化-->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

3.Publisher发送消息给到消息队列

 //Publisher发送消息给到队列object.queue
    @Test
    public void testSendObjectQueue(){
        HashMap<Object, Object> msg = new HashMap<>();
         msg.put("姓名","柳岩");
         msg.put("年龄",38);
        //发送消息到队列
        rabbitTemplate.convertAndSend("object.queue",msg);
    }

在这里插入图片描述

4.消费者接收消息

@RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String,Object> msg){
        System.out.println("消费者接收到object.queue的消息：【"+ msg +"】");
    }

在这里插入图片描述

更多推荐

9.3.5网络原理(应用层HTTP/HTTPS)

一.HTTP:1.HTTP是超文本传输协议,除了传输字符串,还可以传输图片,字体,视频,音频.2.3.HTTP协议报文格式:a.首行,b.请求头(header),c.空行(相当于一个分隔符,分隔了header和body),d.正文(body).4.5.URL:唯一资源描述符(长度不限制).a.b.注意:查询字符串(qu

第29章_瑞萨MCU零基础入门系列教程之改进型环形缓冲区

本教程基于韦东山百问网出的DShanMCU-RA6M5开发板进行编写，需要的同学可以在这里获取：https://item.taobao.com/item.htm?id=728461040949配套资料获取：https://renesas-docs.100ask.net瑞萨MCU零基础入门系列教程汇总：https://b

【初阶数据结构】树（tree）的基本概念——C语言

目录一、树（tree）1.1树的概念及结构1.2树的相关概念1.3树的表示1.4树在实际中的运用（表示文件系统的目录树结构）二、二叉树的概念及结构2.1二叉树的概念2.2现实中真正的二叉树2.3特殊的二叉树2.4二叉树的性质2.5二叉树的存储结构一、树（tree）1.1树的概念及结构树是一种非线性的数据结构，它是由n（

Softing物联网（IoT）方案之OT/IT数据集成

一利用数据提高效率和绩效多年以来数据集成和工业物联网一直在推动着市场的发展，目前我们已经能够集成并成功使用先进的技术、大量的传感器和复杂的数据格式等。而在工业物联网或工业4.0中，还有运营技术（OT）和信息技术（IT）它们之间的无缝数据交换对于企业提升竞争力而言同样非常重要。将生产和业务数据深度集成到IT层中可为新的利

【Oracle】Oracle系列--Oracle数据类型

文章目录前言1.字符类型2.数字类型3.大对象类型4.时间及时间间隔类型5.其他类型前言ORACLE基本数据类型，又叫内置数据类型（built-indatatypes)可以按类型分为：字符串类型、数字类型、大对象类型（LOB类型）、日期类型、LONGRAW&RAW类型、ROWID&UROWID类型。1.字符类型数据类型

浅说 MySQL 数据库日志有哪些？作用是什么？

目录1.MySQL日志有哪些？2.各种日志分析2.1错误日志2.2二进制日志2.3通用查询日志2.4慢查询日志2.5中继日志2.6数据定义语句日志2.7补充点3.日志有什么用(优点)4.日志的弊端1.MySQL日志有哪些？MySQL数据库为我们提供了很多种不同类型的日志文件，用来存储不同类型的日志。主要有错误日志，二进

Redis+SpringBoot企业版集群实战------【华为云版】

目录安装复制及集群bgsaverdbaofSpringBoot+Redis操作操作string操作hash操作set操作sortedset获取所有key&删除设置key的失效时间SpringDataRedis整合使用哨兵机制安装下载地址Redis上传至服务器解压tarzxvfredis-5.0.3.tar.gz安装依赖

JS Ajax 封装

ajax封装一、什么是Ajax？二、Ajax的优缺点？2.1优点2.2缺点三、Ajax的使用3.1状态码3.2xhr的基本使用3.3ajax原生封装：3.3.1触发GET请求：3.3.2调用POST请求：四、Ajax的约束一、什么是Ajax？Ajax（AsynchronousJavaScriptAndXML）是2005

【C++杂货铺】国庆中秋特辑——多态由浅入深详细总结

文章目录一、多态的概念二、多态的定义及实现2.1多态的构成条件2.2虚函数2.3虚函数的重写2.4虚函数重写的两个例外2.4.1协变（基类与派生类虚函数返回值类型不同）2.4.2析构函数的重写（基类与派生类析构函数的名字不同）2.5C++11override和final2.5.1final：修饰虚函数，表示该虚函数不能

ORB-SLAM2_RGBD_DENSE_MAP编译、问题解决、离线加载TUM数据和在线加载D435i相机数据生成稠密地图

文章目录0引言1安装依赖1.1其他库安装1.2pcl库安装2编译ORB-SLAM2_RGBD_DENSE_MAP2.1build.sh2.2build_ros.sh3运行ORB-SLAM2_RGBD_DENSE_MAP3.1build.sh编译版本3.2build_ros.sh编译版本0引言ORB-SLAM2_RGBD

什么是集成测试？集成测试方法有哪些？

1、基本概念：将软件集成起来后进行测试。集成测试又叫子系统测试、组装测试、部件测试等。集成测试主要是针对软件高层设计进行测试，一般来说是以模块和子系统为单位进行测试。2、集成测试包含的层次：1.模块内的集成，主要是测试模块内各个接口间的交互集成关系；2.子系统内的集成，测试子系统内各个模块间的交互关系；3.系统集成，测

热文推荐
  1. 常见的查找算法以及分块搜索算法的简明教程

    顺序查找最基本的查找算法举例//顺序查找publicstaticintsearchSequence(int[]arr,inttarget){inti=0;for(intarr2:arr){if(arr2==target){returni;}i++;}return-1;}二分查找[!warning]值得注意的是这个二分查

  2. Go的性能优化建议

    前言：\textcolor{Green}{前言：}前言：💞这个专栏就专门来记录一下寒假参加的第五期字节跳动训练营💞从这个专栏里面可以迅速获得Go的知识Go的性能优化建议3性能优化建议3.1性能优化建议-Benchmark3.2性能优化建议-slice3.3性能优化建议-Map3.4性能优化建议-字符串处理3.5性能

  3. 【AI视野·今日CV 计算机视觉论文速览 第248期】Mon, 18 Sep 2023

    AI视野·今日CS.CV计算机视觉论文速览Mon,18Sep2023Totally83papers👉上期速览✈更多精彩请移步主页Interesting:📚Robuste-NeRF,处理高速且大噪声事件相机流的NERF模型。(fromNUS新加坡国立)稀疏噪声事件与稠密事件数据的区别：模型架构：项目网站：https:

  4. Python网络编程(socket)

    网络编程指的是：在程序中实现两台计算机之间的通信。Python提供了大量网络编程的工具和库，本文重点学习socket和select模块。网络编程涉及许多关于TCPIP的基础知识，本文默认对这些知识已经了解了，不再对TCPIP相关的知识进行学习。socket模块这个模块提供了访问BSD套接字的接口。在所有现代Unix系统

  5. 开始在 Windows 上将 Python 用于 Web 开发

    🎬岸边的风：个人主页🔥个人专栏:《VUE》《javaScript》⛺️生活的理想，就是为了理想的生活!目录设置开发环境安装适用于Linux的Windows子系统设置VisualStudioCode创建新项目安装Python、pip和venv创建虚拟环境打开WSL-Remote窗口安装MicrosoftPython扩

  6. 【C# 基础精讲】抽象类与接口

    抽象类（AbstractClass）和接口（Interface）是面向对象编程中两种重要的概念，它们用于定义类的结构、行为和关系，是实现多态性、代码复用和系统设计的关键手段。在C#及其他面向对象编程语言中，抽象类和接口都发挥着重要作用。本文将详细解释抽象类和接口的概念、特点、用法以及在C#中的应用。1.抽象类的概念与特

  7. 同城信息服务源码 本地生活服务小程序源码

    同城信息服务源码本地生活服务小程序源码功能介绍：基本设置：网站参数、安全设置、分站管理、支付设置、操作日志、地区设置、公交地铁、国际区号、清理缓存、模板风格、模块管理、域名管理、底部菜单、消息通知、登录设置其他设置：关键词管理、单页文档、网站公告、帮助信息、网站协议、广告设置、友情链接、举报管理、意见反馈、投诉列表短信

  8. CH573-09-BLE蓝牙安卓应用二次开发——RISC-V内核BLE MCU快速开发教程

    一、基础工程搭建在上一章最后一讲的BLE蓝牙例程中，我们使用了沁恒官方的BLE调试助手完成数据发送，接下来我们使用AndroidStudio完成一款简易的BLE调试助手。1、参考文章我这里参考了CSDN中的一位博主“摸爬滚打的程序媛”的文章以及对应文章中的AndroidStudioBLE应用工程的Demo。版权声明：链

  9. MP3算法及代码例程

    MP3（MPEG-1AudioLayerIII）是一种数字音频压缩算法，用于对音频进行高效的压缩。MP3算法能够显著减小音频文件的大小，同时保持较高的音质。以下是MP3算法的主要步骤：采样率转换：将输入音频信号的采样率转换为固定的值，通常为44.1kHz。这是因为人耳对于音频的感知范围大约在20Hz到20kHz之间，因