RabbitMQ的介绍和使用

1.同步通讯和异步通讯

        举个例子,同步通讯就像是在打电话,因此它时效性较强,可以立即得到结果,但如果你正在和一个MM打电话,其他MM找你的话,你们之间是不能进行消息的传递和响应的

        异步通讯就像是微信,你可以和多个MM进行消息的传递,对方可以立即响应或者有空了在”回“你

        同步调用问题

        微服务间基于Feign的调用就属于同步方式,存在一些问题。

        1.代码耦合:如果后续需要添加业务,需要不断修改支付服务的代码

        2.性能下降,吞吐量下降:支付服务一直在等待,订单等服务的响应,cpu资源一直在占用。

        3.级联失败:如果服务提供者出现问题,所有调用方都会跟着出问题。

2.异步调用方案

        异步调用常见的就是事件驱动模式:Broker是事件代理者,一旦用户支付成功,这就是一个事件,这个事件就会被Broker管理,订单服务、仓储服务、短信服务。就会找Broker,这个叫做订阅事件。一旦用户支付成功之后,Broker就会通知被订阅过事件的服务,支付服务完成事件发布之后,就结束了服务,返回给用户

        优点:

        解决代码解耦:添加业务时不需要再更改支付服务的代码,支付服务只需要发布事件就行。至于后面的业务支付服务可以不用考虑。

        性能提升,吞吐量提供:相比较同步服务,业务处理时间的累加,支付服务还需要等待其他服务完成并响应,通过异步的方式,支付服务发布时间之后就结束服务,无需等待其他服务响应。提升了性能,和吞吐量

        服务没有依赖关系,不用担心级联失败问题

        流量削峰: 有多个事件发布,可以囤积到broker上,订阅该事件的服务可以按自己的处理能力来稳步进行。broker起到缓冲作用

        缺点:

        依赖Broker的可靠性、安全性、吞吐能力

        架构复杂了,业务没有明显的流程线,不好追踪管理

3.什么是MQ

MQ(MessageQueue),中文是消息队列,字面来看就是存放消息的队列。也就是事件驱动架构中的Broker

常见的MQ

RabbitMQ,适用于中小型企业开发,如果对性能要求比较高的并且需要定制服务的大型企业推荐使用Kafka。下面会介绍RabbitMQ的使用。

4.RabbitMQ概述和安装

RabbitMQ概述

RabbitMQ是基于Erlang语言开发的开源消息通信中间件,官网地址:https://www.rabbitmq.com

RabbitMQ的部署

环境:centos7,docker在线拉取的方式部署。

#拉取RabbitMQ镜像

输入:docker pull rabbitmq:3-management

#设置默认用户名密码并启动容器

docker run --name rabbitmq -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=123 -p 15672:15672 -p 5672:5672 -d rabbitmq:3-management

端口15672是rabbitMQ的ui界面,5672是服务端口

ui界面展示:

RabbitMQ结构和概念

PubLisher是我们的消息(事件)发送者,consumer是消息的消费者,PubLisher将来会把消息发送到我们的exchange(交换机)上,交换机负责路由,并把消息投射到queue(队列),queue负责暂存消息,consumer负责从queue里面获取消息处理消息。

 

5.RabbitMQ的常见消息模型

一、基本消息队列(BasicQueue)

HelloWorld是最基本的消息队列模型,实现的话,包含三个角色

publisher:消息发布者,将消息发送到队列queue

queue:消息队列,负责接受并缓存消息

consumer:订阅队列,处理队列中的消息

官方提供的编码方式非常麻烦,下面我们介绍学习一下SpringAMQP,它可以大大简化我们消息发送和接收API。

SpringAMQP简介

AMQP:是用于在应用程序或之间传递业务消息的开放标准,该协议与语言的平台无关,更符合微服务中独立性的要求。

Spring AMQP:是基于AMQP协议定义的一套API规范,提供了模板来发送和接收消息,包含两部分,其中spring-amqp是基础抽象,spring-rabbit是底层的默认实现。

利用SpringAMQP实现基础消息队列功能

通过rabbitTemplate提供的convertAndSend就可以实现消息的发送

引入相关依赖

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <scope>test</scope>
</dependency>

publisher(消息发布者)的application.yml的配置

test的测试代码:

@RunWith(SpringRunner.class)
@SpringBootTest
class PublisherApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
        String queuename="simple queue";
        String message="hello,spring AMQP";
        rabbitTemplate.convertAndSend(queuename,message);
    }

}

通过rabbitTemplate实现对队列消息的监听

引入依赖

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <scope>test</scope>
</dependency>

配置consumer(消息接收者),的application.yml文件

spring:
  rabbitmq:
    host: 192.168.10.8 #主机名
    port: 5672         #端口
    username: root      #用户名
    password: 123       #密码
    virtual-host: /     #虚拟主机名

监听类的代码

@Component
public class SimpleListener {
    @RabbitListener(queues = "simple queue")
    public void ListenSimpleQueue(String msg)
    {
        System.out.println("消费者接收到simple queue的消息:"+msg);
    }
}

ps:消息一旦消费就会从队列中删除,RabbitMQ没有消息回溯功能。

二、工作消息队列(WorkQueue)

工作队列的结构如下:

工作消息队列的结构,相比于基础消息队列多了个消费者,因为rabbitMQ阅后即焚的特性这两个消费者属于共同工作的关系,如果有50个消息,他们两个消费者就会一人分一半,也就是一人25条,为啥会多一个消费者?这是因为如果一个消费者每次处理40个消息,但是publisher一次发布50个消息,多出来的消息会存储在queue里面,又因为queue是占用内存的假以时日,内存就会爆满,新的消息就存不进去了,多一个消费者每次就可以处理80条消息,可以有效解决这个问题。

work queue,工作队列,可以提高消费处理速度,避免队列消息堆积。但是这里有一个消息预取机制 ,消费者会提前把消息拿过来,因此消息是平局分配,并不是“能者多劳”的模式,通过设置prefetch的值来实现每次只能获取一条消息,处理完成才能接取下一个消息。实现“能者多劳”的模式。

  1. 发布订阅(Publish、Subscribe)

基础消息队列和工作消息队列都是一条信息只被一个消费者消费,消费完就删除,显然不能实现我们之前预想的完成支付之后,通知仓储、短信等服务。这就需要我们了解学习发布订阅模式。发布订阅模式与之前案例的区别就是允许同一消息发送给多个消费者,实现方式是加入exchange(交换机),结构如下:

publisher将消息发送给exchange(交换机),交换机把这个消息转发给队列,因此,发布者(publisher)并不需要知到转发给了那个 队列或多个队列,转发给多个队列,这种方式就能实现被多个消费者消费,那么交换机到底是发给一个还是多个呢?这是由交换机类型来决定的。常见的exchange的类型包括:

广播:Fanout

路由:Direct

主题:Topic

注意: exchange负责消息路由,而不是储存,路由失败则消息丢失。

广播-Fanout Exchange

Fanout Exchange 会将接收到的消息路由到每一个绑定的queue。

实现思路:

1.在consumer服务中,利用代码声明队列、交换机、并将两者绑定。

在consumer服务上添加@Configuration注解,并声明FanoutExchange、Queue和关系对象Binding,代码如下:

@Configuration
public class FanoutConfig {
    //声明交换机对象
    @Bean
    public FanoutExchange fanoutExchange()
    {
        return new FanoutExchange("fanout");
    }
    //声明队列1
    @Bean
    public Queue fanoutqueue1()
    {
        return new Queue("fanoutqueue1");
    }
    //绑定队列一到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutqueue1,FanoutExchange fanoutExchange)
    {
        return BindingBuilder.bind(fanoutqueue1).to(fanoutExchange);
    }
    //声明队列2
    @Bean
    public Queue fanoutqueue2()
    {
        return new Queue("fanoutqueue2");
    }
    //绑定队列二到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutqueue2,FanoutExchange fanoutExchange)
    {
        return BindingBuilder.bind(fanoutqueue2).to(fanoutExchange);
    }
}

2.在consumer服务中,编写两个消费者方法,分别监听fanout.queue1和fanout.queue2

@RabbitListener(queues = "fanoutqueue1")
public void ListenSimpleQueue3(String msg) throws InterruptedException {
    System.out.println("消费者222接收到fanoutqueue1的消息:"+msg);
    Thread.sleep(100);
}
@RabbitListener(queues = "fanoutqueue2")
public void ListenSimpleQueue4(String msg) throws InterruptedException {
    System.out.println("消费者222接收到fanoutqueue2的消息:"+msg);
    Thread.sleep(100);
}

3.在publisher中编写测试方法,向fanout发送消息。

@Test
public void contectFonoutExchange()
{
    //交换机名称
    String exchangeName="fanout";
    //消息内容
    String message="hello everyone";
    //发送消息
    rabbitTemplate.convertAndSend(exchangeName,"",message);
}

路由-DirectExchange

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue,因此称为路由模式(routes)

每一个Queue都与Exchange设置一个BindingKey,一个队列可以绑定多个BindingKey。

发布者发送消息时,指定消息的RoutingKey

Exchange(交换机)将消息路由到Bindingkey与消息RoutingKey一致的队列

实现思路:

1.利用@RabbitListenner声明Exchange、Queue、RountingKey

2.在consumer服务中编写两个消费者方法,分别监听queue1和queue2

在我们的监听类里面增加两个方法,用来声明交换机、队列和RountingKey

//发布订阅DirectExchange
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "queue1"),
        exchange = @Exchange(name = "direct",type= ExchangeTypes.DIRECT),
        key = {"red","blue"}
))
public void listenerDirectqueue1(String msg)
{
    System.out.println("消费者接收到direct.queue1的消息"+msg);
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "queue2"),
        exchange = @Exchange(name = "direct",type= ExchangeTypes.DIRECT),
        key = {"red","yellow"}
))
public void listenerDirectqueue2(String msg)
{
    System.out.println("消费者接收到direct.queue2的消息"+msg);
}

3.在publisher中编写测试方法,向Exchange发送消息。

@Test
public void contextDirectExchange()
{
    String exchangeName="direct";
    String msgblue="hello blue";
    String msgRed="hello red";
    String msgYellow="hello yellow";
    rabbitTemplate.convertAndSend(exchangeName,"blue",msgblue);
}

@Test
public void contextDirectExchange()
{
    String exchangeName="direct";
    String msgblue="hello blue";
    String msgRed="hello red";
    String msgYellow="hello yellow";
    rabbitTemplate.convertAndSend(exchangeName,"red",msgRed);
}

话题-TopicExchange

TopicExchange与DirectExchange类似,区别在于rountingKey必须是多个单词的列表,并且以"."分割。Queue与Exchange指定BindingKey可以使用通配符

#:代表0个或多个单词

*:代表一个单词

我们使用Direct的时候一个队列如果绑定了很多key,会非常麻烦,通配符的引入就把key的绑定简化许多,原来绑定多个key现在只需要绑定一个key。

实现思路:

1.利用@RabbitListenter声明Exchange、Queue、RoutingKey

2.在consumer服务中,编写两个消费者方法,分别监听topic.queue1和topic.queue2

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

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

3.在publisher中编写测试方法,向交换机topic发送消息

@Test
public void contextTopicExchange()
{
    String exchangeName="topic";
    String msg="我是懒大王";
    rabbitTemplate.convertAndSend(exchangeName,"china.news",msg);
}

6.消息转换器

在SpringAMQP的发送方法中,接收消息的类型是Object,也就是说我们可以发送任意对象类型的消息,SpringAMQP会帮我们序列化为字节后发送。

Spring的消息对象处理是由MessageConcerter来处理的,而默认实现是SimpleMessageConverter,基于JDK的ObjectOutputStream完成序列化,这种序列化方式,比较浪费内存资源,如果需要修改,只需要定义一个MessageConverter类型的Bean即可。推荐用JSON方式序列化,步骤如下:

我们在publisher服务引入依赖

我们在publisher服务中声明MessageConverter

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/599901.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

Linux系统使用Docker安装青龙面板并实现远程访问管理面板

文章目录 一、前期准备本教程环境为&#xff1a;Centos7&#xff0c;可以跑Docker的系统都可以使用。本教程使用Docker部署青龙&#xff0c;如何安装Docker详见&#xff1a; 二、安装青龙面板三、映射本地部署的青龙面板至公网四、使用固定公网地址访问本地部署的青龙面板 青龙…

【busybox记录】【shell指令】shuf

目录 内容来源&#xff1a; 【GUN】【shuf】指令介绍 【busybox】【shuf】指令介绍 【linux】【shuf】指令介绍 使用示例&#xff1a; 打乱内容 - 默认输出 打乱内容 - 最多输出n行 打乱内容 - 将输出写入文件 打乱内容 - 重复输出 打乱内容 - 打乱本条指令的参数 打…

C语言 自定义类型——联合体

目录: 一、联合体是&#xff1f;声明计算内存大小 二、联合体的特点例如 三、联合体大小的计算规则&#xff1a; 四、应用习1习2 一、联合体是&#xff1f; 联合体和结构体差不多&#xff0c;但是其最大的区别在于联合体所有的成员共用一块内存空间。所以联合体也叫共用体。联…

方法的重写

方法的重写 概念&#xff1a;子类继承父类之后&#xff0c;就拥有了符合权限的父类的属性和方法&#xff0c;但是当父类的方法不符合子类的要求的时候&#xff0c;子类也可以重新的书写自己想要的方法。所以&#xff0c;方法的重写&#xff0c;即子类继承父类的方法后&#xf…

「代码与养生」 :当下程序员的养生指南

前言 众所周知&#xff0c;程序员是死的比较快的一类人。因为天天加班、睡眠不足、久坐不动、长时间面对电子屏幕辐射、长时间高强度用脑等不好避免的问题。因此&#xff0c;要想活的时间长一点&#xff0c;就要多了解一些养生之道 下面&#xff0c;根据个人看的一些博客、书…

深度学习之基于Matlab卷积神经网络验证码识别系统

欢迎大家点赞、收藏、关注、评论啦 &#xff0c;由于篇幅有限&#xff0c;只展示了部分核心代码。 文章目录 一项目简介 二、功能三、系统四. 总结 一项目简介 一、项目背景 随着互联网的发展&#xff0c;验证码作为一种常用的安全验证手段&#xff0c;被广泛应用于各种网站和…

网络安全--红队资源大合集

目录 相关资源列表 攻防测试手册 内网安全文档 学习靶场 工具包集合 内网安全文档 学习手册相关资源 产品设计文档 版本管理平台漏洞收集 相关工具拓展插件 Kali 环境下拓展插件 Nessus 相关工具拓展插件 Awvs 相关工具拓展插件 红队攻击的生命周期&#xff0c;…

基于FPGA的AD7705芯片驱动设计VHDL代码Quartus仿真

名称&#xff1a; 软件&#xff1a;Quartus基于FPGA的AD7705芯片驱动设计VHDL代码Quartus仿真&#xff08;文末获取&#xff09; 语言&#xff1a;VHDL 代码功能&#xff1a; AD77025芯片控制及串口输出 1、使用FPGA控制AD77025芯片&#xff0c;使其输出AD值 2、将数据计…

基于springboot实现夕阳红公寓管理系统项目【项目源码+论文说明】

基于springboot实现夕阳红公寓管理系统演示 摘要 如今社会上各行各业&#xff0c;都在用属于自己专用的软件来进行工作&#xff0c;互联网发展到这个时候&#xff0c;人们已经发现离不开了互联网。互联网的发展&#xff0c;离不开一些新的技术&#xff0c;而新技术的产生往往是…

ASP.NET网上鲜花销售系统的设计

摘 要 本系统实现了一般电子商务所具备的功能&#xff0c;如商品浏览、用户登录注册、网上与购物、结算、后台数据库管理等&#xff0c;利用这些功能可以对鲜花销售信息进行较好的管理。 网上鲜花销售系统的使用者主要是客户和销售管理者&#xff0c;对于客户来说&#xff0…

欧式聚类提取-------PCL

欧式聚类 std::vector<pcl::PointCloud<pcl::PointXYZ>::Ptr> PclTool::euclideanClustering(const pcl::PointCloud<pcl::PointXYZ>::Ptr& cloud) {std::vector<pcl::PointCloud<pcl::PointXYZ>::Ptr> clustered_clouds;// 下采样pcl::Vox…

基于FPGA的DDS波形发生器VHDL代码Quartus仿真

名称&#xff1a;基于FPGA的DDS波形发生器VHDL代码Quartus仿真&#xff08;文末获取&#xff09; 软件&#xff1a;Quartus 语言&#xff1a;VHDL 代码功能&#xff1a; DDS波形发生器VHDL 1、可以输出正弦波、方波、三角波 2、可以控制输出波形的频率 DDS波形发生器原理…

学习和分析各种数据结构所要掌握的一个重要知识——CPU的缓存利用率(命中率)

什么是CPU缓存利用率&#xff08;命中率&#xff09;&#xff0c;我们首先要把内存搞清楚。 硬盘是什么&#xff0c;内存是什么&#xff0c;高速缓存是什么&#xff0c;寄存器又是什么&#xff1f; 我们要储存数据就要运用到上面的东西。首先里面的硬盘是可以无电存储的&#…

无人零售,重塑购物新纪元

在这个快节奏的时代&#xff0c;科技的每一次跃进都在悄无声息地改变着我们的生活方式。而今&#xff0c;无人零售正以雷霆之势&#xff0c;颠覆传统购物模式&#xff0c;为我们带来前所未有的便捷与智能体验。想知道无人零售如何彻底改变我们的购物方式吗&#xff1f;跟随我&a…

Linux实现Flappy bird项目

目录 1、项目介绍 2、功能总结 3、前期准备 3.1 Ncurses库 3.2 信号机制 3.2.1 设置信号响应方式 3.2.2 设置定时器 4、代码实现 4.1 头文件引用及变量、函数定义 4.2 主函数 4.3 curses初始化 4.4 设置定时器 4.5 定时器响应函数 4.6 小鸟控制相关函数 4…

Java的Fork-Join简单介绍

Java的Fork-Join框架是Java 7引入的一个用于并行处理的轻量级框架&#xff0c;它基于分治策略&#xff08;Divide and Conquer&#xff09;&#xff0c;特别适合于那些可以被分解为多个子任务的任务。Fork-Join框架的核心思想是将一个大任务&#xff08;Task&#xff09;拆分成…

毕业就业信息|基于Springboot+vue的毕业就业信息管理系统的设计与实现(源码+数据库+文档)

毕业就业信息管理系统 目录 基于Springboot&#xff0b;vue的毕业就业信息管理系统设计与实现 一、前言 二、系统设计 三、系统功能设计 1学生信息管理 2 公司信息管理 3公告类型管理 4公告信息管理 四、数据库设计 五、核心代码 六、论文参考 七、最新计算机毕设…

高等数学笔记(下中)

曲线积分 第一类曲线积分&#xff1a;对弧长的积分计算方法 定理&#xff1a;设 f ( x , y ) f(x,y) f(x,y)在曲线弧 L L L上有定义且连续&#xff0c; L L L的参数方程是 { x φ ( t ) y ψ ( t ) ( α ≤ t ≤ β ) \begin{cases} x\varphi(t)\\ y\psi(t) \end{cases}(\a…

Python实现打砖块游戏

提供学习或者毕业设计使用&#xff0c;功能基本都有&#xff0c;不能和市场上正式游戏相提比论&#xff0c;请理性对待&#xff01; 在本文中&#xff0c;我们将使用 Pygame 和 Tkinter 创建一个简单的打砖块游戏。游戏的目标是通过控制挡板来击碎屏幕上的砖块&#xff0c;同时…

PTQ4SAM、Mamba-Attention、AniTalker、IceFormer、U-DiTs、CogDPM

本文首发于公众号&#xff1a;机器感知 PTQ4SAM、Mamba-Attention、AniTalker、IceFormer、U-DiTs、CogDPM PTQ4SAM: Post-Training Quantization for Segment Anything Segment Anything Model (SAM) has achieved impressive performance in many computer vision tasks. Ho…
最新文章