RabbbitMQ基本使用及其五种工作模型

初识MQ

同步通讯和异步通讯

        什么是同步通讯呢?举个例子,你认识了一个小姐姐,聊的很火热,于是你们慢慢开始打电话,视频聊天,这种方式就成为同步通讯,那什么是一部通讯呢,同样的,你认识了多个小姐姐,你和他们进行文字聊天,这时候你一个人可以和多个人聊天,这就是异步通讯。我们之前进行服务间调用时使用的RestTemplaste,Feign就是同步调用。

同步调用的优缺点

        优点:时效性强,可以立即得到回复,就像你打视频一样

        缺点:假如你一个项目中存在很多业务,相互之间进行调用,如果你增加了新的需求,此时因为原本代码是同步调用,代码耦合度很高,于是乎修改代码变得十分繁琐,并且一个业务可能会调用很多服务,只有上一个服务调用完了,才能到下一个服务,等待的过程中就造成了资源浪费,性能下降,如果当前调用的服务失败,还可能会导致级联失败,服务雪崩。


异步调用的优缺点

优点:

1,代码耦合度低,因为异步调用是采取事件驱动来实现,当请求进来之后到达Broker之后Broker通知各自微服务去执行,服务间不在需要相互调用。

2,吞吐量提升,因为异步调用不像同步调用那样每个服务需要等待上游完成调用。

3,故障隔离,服务之间相互不进行调用,即使你挂了也跟我没关系。

4,流量削峰,假如同时又大量请求,但是你的服务处理请求能力是有限的,于是Broker会净请求先拦截,看服务又能力处理多少请求,就拿多少请求,不会一次性全部发布订阅。

缺点:

1,对Broker的依赖十分高,对他的可靠性,安全性,吞吐能力要求很高,万一他挂了......

2,服务之间相互调用关系不清晰,业务没有明显的流程线,代码出问题不容易排查。

所以,需要根据场景来选择同步还是异步,一般大多数需要同步。

什么是MQ?

MQ(MessageQueue),中文是消息队列,也就是存放消息的队列,也就是时间驱动中的Broker.

常见的MQ对比:

我们这里选择RabbitMQ

RabbitMQ安装

RabbitMQ是基于Erlang语言开发的开源消息中间件,因此它具备强大的并发处理能力

官网地址:RabbitMQ: One broker to queue them all | RabbitMQicon-default.png?t=N7T8https://www.rabbitmq.com/

这里我们只用Docker来安装RabbitMQ(快捷方便)

1,拉取RabbitMQ的镜像

docker pull rabbitmq:3-management 

2,运行RabbitMQ容器

docker run \
 -e RABBITMQ_DEFAULT_USER=你的账户名\
 -e RABBITMQ_DEFAULT_PASS=你的密码\
 --name mq \  
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

注意:账户名和密码是自己定义的,15672是RabbitMQ的管理端端口,5672是RabbitMQ通讯的端口。

3,在浏览器输入IP地址:15672,输入帐号名和密码登录

至此,安装成功

我们可以看到界面有好多目录,具体作用如下

channel:用来操作mq的工具

exchange:路由消息到队列中

queue:缓存消息

virtual host:虚拟主机,是对queue,exchange等资源的逻辑分组

MQ的整体结构:

消息发送者将消息发送到交换机,交换机将其路由到队列,消费者从队列中取走消息

MQ中常见消息模型

大致可以分为两类

第一类:基本消息队列(BasicQueue),工作消息队列(WorkQueue),他们都是最基本的消息队列

第二类:发布订阅(publish,Subscribe)根据交换机类型分为三种

Fanout Exchange(广播),Direct Exchange(路由),Topic Exchange(主题)

  

更多可参考官方文档: 

RabbitMQ Tutorials | RabbitMQ

RabbitMQ入门案例

我们使用RabbitMQ参考官方文档,完成一个hello world案例

引入依赖:

     <!--包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

创建一个测试类publisher,用来发送消息代码如下:

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.*.*");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("***");
        factory.setPassword("****");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
        // 2.创建通道Channel
        Channel channel = connection.createChannel();
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功:【" + message + "】");
        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}

Debug执行观察:

1,连接工厂初始化完成之后,他会创建一个连接,连接上RabbitMQ ,这时界面显示如下

2,之后他会创建一个Channel ,用于操作RabbbitMQ

3,之后会创建一个我们定义好的消息队列simple.queue:

4,之后发送消息hello,rabbitmq!

5,完成之后关闭连接,但是消息依旧存在与消息队列中

之后创建一个测试类consumer,用来接受消息代码如下:

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.*.*");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("****");
        factory.setPassword("****");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
        // 2.创建通道Channel
        Channel channel = connection.createChannel();
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息:【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

 跟publisher一样创建工厂,建立连接,这里需要说明的是,之所以还要创建一个队列是因为在实际执行过程中,发布者和消费者又可能执行顺序不一致,所以我们消费者也需要创建一个队列,不过这个队列只会有一个,如果创建了就不再创建。之后消费者接受消息处理,消息队列中消息清空。

根据上述代码我们可以看到官方的demo确实是有带你复杂繁琐,实在是很不友好啊,于是我们使用一种简单的方式来操作RabbitMQ!


SpringAMQP

AMQP(Advanced Message Queuing Protocal):高级消息队列协议,是应用程序之间传递业务消息的开放标准/规范,和语言和平台无关。

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

SpringAMQP实现基础消息队列功能

消息发送

1,引入依赖

  <!--AMQP依赖,包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

2,在配置文件中配置RabbitMQ的信息

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

3,使用RabbitTemplate来发送消息(spring提供的发送消息的模板)

这里我们编写一个测试类来测试

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

    @Test
    public void sendMessage(){
        String queueName="simple.queue";//定义队列名
        String message="hello SpringAMQP!!";
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

 可以看到消息成功发送

消息接收

依旧是先引依赖,添加RabbitMQ相关配置

之后书写监听消息代码,创建一个类,交给spring管理,定义一个方法加@RabbitListener注解指定接受消息的队列,可以传递数组,指定多个队列

@Component
public class ListenerMessage {
    @RabbitListener(queues = {"simple.queue"})//可以监听多个队列
    public void ListenerSimpleQueue(String message){
        System.out.println("接收到的消息是;"+message);
    }
}

可以看到成功接受消息

 WordQueue 工作队列

Work模型-多个消费者绑定到同一个队列,同一个消息只会被同一个消费者处理

我们来做一个测试,在一秒内发送50条消息,定义两个消费者,同时处理,一个消费者每秒处理50条消息,另一个消费者每秒处理20个消息,按照常理来说应该是多劳多得,也就是处理能力强的处理更多消息,弱的处理更少消息,我们修改之前的代码:

修改之后的消息发送代码:

@Test
    public void sendWorkMessage() throws InterruptedException {
        String queueName="simple.queue";//定义队列名
        String message="Work Message__";
        for (int i = 1; i <= 50; i++) {
            rabbitTemplate.convertAndSend(queueName,message+i);
            Thread.sleep(20);
        }
    }

修改之后的消息接收代码:

@Component
public class ListenerMessage {

    @RabbitListener(queues = {"simple.queue"})//可以监听多个队列
    public void ListenerWorkSimpleQueue1(String message) throws InterruptedException {
        System.out.println("consumer1接收到的消息是;"+message+"  时间:"+LocalDateTime.now());
        Thread.sleep(20);
    }
    @RabbitListener(queues = {"simple.queue"})//可以监听多个队列
    public void ListenerWorkSimpleQueue2(String message) throws InterruptedException {
        System.err.println("consumer2接收到的消息是;"+message+"  时间:"+LocalDateTime.now());
        Thread.sleep(200);
    }
}

代码执行结果显示,consumer1和consumer2处理的消息是一样的并不会向我们设想的那样,能力强的处理多能力弱的处理少,是什么原因导致的呢??

这是由于RabbitMQ的消息预取机制,就是说在消息到达消息队列的时候两个消费者会分别从消息队列中一次性取完所有的消息理论上来说是无上限的,所以我们需要修改机制让消费者一次性例如说取一个消息,等这个消息处理完之后在去取下一个即可,

在配置文件中修改如下:

spring:
  rabbitmq:
    host: 192.168.101.100 #主机名
    port: 5672 #端口
    virtual-host: /  #虚拟主机
    username: qmlx #用户名
    password: QMLX-9999 #密码
    listener:
      direct:
        prefetch: 1  #每次取一个消息,取完之后在取

 发布(Publish)订阅(Subscribe)

发布订阅模式和之前案例的区别就是上述模型一个消息只能发送给一个consumer,而发布订阅模型则是将同一个消息发送给多个消费者,实现方式是假如exchange(交换机)

 

常见的exchange类型包括:

Fanout:广播

Direct:路由

Topic:话题

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

🧇发布订阅-Fanout Exchange

Fanout Exchange他会将接收到的消息路由到每一个与其绑定的queue,广播模式。

实现思路:

1,在consumer服务中,编写一个配置类,声明两个队列,一个交换机,并将交换机绑定到队列上


@Configuration
public class FanoutConfig {
    @Bean
    public FanoutExchange fanoutExchange(){
        return  new FanoutExchange("fanout.exchange");
    }

    //定义两个队列
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    //将队列绑定在交换机上面
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
    //按照类型和名称传入参数
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }


}

启动项目之后spring会自动加载交换机和队列

 2,编写consumer代码,监听定义好的两个队列


@Component
public class ListenerMessage {

     @RabbitListener(queues = {"fanout.queue1"})
    public void ListenerFanoutQueue1(String message){
        System.out.println("consumer1接受到的消息是:"+message);
    }

    @RabbitListener(queues = {"fanout.queue2"})
    public void ListenerFanoutQueue2(String message){
        System.err.println("consumer2接受到的消息是:"+message);
    }
}

3,编写publisher代码,发送消息

之前消息是发送到队列中,现在消息发送给交换机

    @Test
    public void sendFanoutMessage() throws InterruptedException {
        String exchangeName="fanout.exchange";//定义队列名
        String message="FanoutExchange Message!!";
        rabbitTemplate.convertAndSend(exchangeName,null,message);

    }

可以看到两个消费者同时接受到了消息!!!

交换机的作用:

1,接受publisher发送的消息

2,将消息按照规则路由发送给每一个与之绑定的队列

3,不能缓存信息,路由失败则消息丢失

🧇发布订阅-Direct Exchange

Direct Exchange:交换机将接收到的消息根据路由规则到指定的Queue,称之为路由模式(routes)

        每一个Queue都和Exchange设定一个BindingKey

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

        Exchange将消息路由到BindKey和消息的BindingKey一致的队列

实现思路:

1,指定消息接收者绑定交换机和队列,但是需要指定BindingKey

这次直接使用注解实现,不必那莫繁琐

@Component
public class ListenerDirectMessage {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "direct.exchange",type = ExchangeTypes.DIRECT),
            key = {"red","blue"}
    ))
    public void listensterDirectQueue1(String message){
        System.out.println("消费者1接收到Direct的消息是:"+message);
    }

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

‘2,定义publisher发布消息

@Test
    public void sendDirectMessage() throws InterruptedException {
        String exchangeName="direct.exchange";//定义队列名
        String message="DirectExchange Message!!";
        String routinfKey="blue";
        rabbitTemplate.convertAndSend(exchangeName,routinfKey,message);
        rabbitTemplate.convertAndSend(exchangeName,null,message);

    }

发送时需要指定routingKey即可,同一个队列定义时可指定多个BindingKey

注意:Direct Exchange只会发送给routingKeyBindingKey一致的队列

🧇发布订阅-Topic Exchange

Topic Exchange:Topic Exchange和Direct key类似,区别在于他的routingKey必须是多个单词的列表,并且必须是以 分割,并且Queue和Exchange指定时课使用通配符

实现思路:

1,实现消费者代码


@Component
public class ListenerTopicMessage {

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

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "topic.exchange",type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listensterDirectQueue2(String message){
        System.out.println("消费者2接收到topic的消息是:"+message);
    }
}

2,实现publisher代码

@Test
    public void sendtopictMessage() throws InterruptedException {
        String exchangeName="topic.exchange";//定义队列名
        String message="TopicExchange Message!!";
        String routinfKey="china.news";
        rabbitTemplate.convertAndSend(exchangeName,routinfKey,message);
        //rabbitTemplate.convertAndSend(exchangeName,null,message);

    }

至此RabbitMQ的安装使用及其五种基本工作模式搞定!!!!

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

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

相关文章

gitlab(docker)安装及使用

GitLab GitLab 是一个用于仓库管理系统的开源项目&#xff0c;使用Git作为代码管理工具&#xff0c;并在此基础上搭建起来的Web服务。 下载(docker) 查询docker镜像gitlab-ce gitlab-ce是它的社区版 [rootlocalhost ~]# docker search gitlab-ce NAME …

OpenCV基本图像处理操作(六)——直方图与模版匹配

直方图 cv2.calcHist(images,channels,mask,histSize,ranges) images: 原图像图像格式为 uint8 或 float32。当传入函数时应 用中括号 [] 括来例如[img]channels: 同样用中括号括来它会告函数我们统幅图 像的直方图。如果入图像是灰度图它的值就是 [0]如果是彩色图像 的传入的…

SETR——Rethinking系列工作,展示使用纯transformer在语义分割任务上是可行的,但需要很强的训练技巧

题目:Rethinking Semantic Segmentation from a Sequence-to-Sequence Perspective with Transformers 作者: 开源:https://fudan-zvg.github.io/SETR 1.研究背景 1.1 为什么要研究这个问题? 自[ 36 ]的开创性工作以来,现有的语义分割模型主要是**基于全卷积网络( FCN )的…

ubuntu20.04安装+ros-noetic安装+内网穿透frp

刷机后的系统安装 ubuntu20.04安装安装ros-noetic安装各种必要的插件安装vscode内网穿透连接实验室主机配置frpc和frps文件运行完成自动化部署免密登录linux的免密登录windows上的免密登录 内网穿透的参考链接&#xff1a;如何优雅地访问远程主机&#xff1f;SSH与frp内网穿透配…

Python学习笔记 - 正则表达式

前言 正则表达式&#xff08;Regular Expression&#xff0c;在代码中常简写为 regex、regexp、RE 或 re&#xff09;是预先定义好的一个“规则字符串”&#xff0c;通过这个“规则字符串”可以匹配、查找、替换那些符合“规则”的文本&#xff0c;也就是说正则表达式针对的目标…

MSTP/RSTP的保护功能

目录 原理概述 实验目的 实验内容 实验拓扑 1.配置RSTP/MSTP 2.配置BPDU保护 3.配置根保护 4.配置环路保护 5.配置TC-BPDU保护 原理概述 在RSTP或MSTP交换网络中&#xff0c;为了防止恶意攻击或临时环路的产生&#xff0c;可配置保护功能来增强网络的健壮性和安全性。…

C++vector类(个人笔记)

vector类 1.熟悉vector接口以及使用1.1vector的定义1.2vector迭代器使用1.3vector空间增长1.4vector增删查改1.5vector迭代器失效问题&#xff08;重点&#xff09; 2.vector的一些笔试题3.模拟实现vector 1.熟悉vector接口以及使用 vector的C官网文档 1.1vector的定义 (con…

用python快速读取大文件几个GB以上的csv数据文件

用python快速读取大文件几个GB以上的csv数据文件 遇到几个GB的csv大文件,用python读取时,可以通过next()函数一行行来读取以提高效率,然后分批量进行处理。 1、文件格式例图 其中第一、第二行是数据行数、列数汇总。 2、流程 1、把csv第一、第二行的数据说明,先读取出来…

Windows远程桌面连接虚拟机Linux

Windows远程桌面连接虚拟机Linux 需要先打开虚拟机的启用VNC连接使用VNC客户端进行连接 yum install -y tigervnc-server #安装tigervnc-server vncserver #启动一个vnc进程 #第一次启动会要求设置密码 #如果需要更改密码可以使用vncpasswd进行更改密码 vncserver -list #查看…

ASUS华硕ROG幻13笔记本电脑GV301R工厂模式原厂OEM预装Windows11系统,恢复出厂开箱状态

适用于型号&#xff1a;GV301RC、GV301RE、GV301RA 工厂模式安装包&#xff1a;https://pan.baidu.com/s/1gLme1VqidpUjCLocgm5ajQ?pwddnbk 提取码&#xff1a;dnbk 工厂模式Win11安装包带有ASUS RECOVERY恢复功能、自带所有驱动、出厂主题壁纸、系统属性专属联机支持标志…

java算法day55 | 动态规划part16 ● 583. 两个字符串的删除操作 ● 72. 编辑距离

583. 两个字符串的删除操作 思路&#xff1a; 和1143.最长公共子序列这道题思路相同&#xff0c;只不过需要对return的数据做一些操作。 class Solution {public int minDistance(String word1, String word2) {int[][] dpnew int[word1.length()1][word2.length()1];for(int …

06_定时器中断

72分频 72MHz 72000000 经过72分频 1000000

【攻防世界】ics-07

<?php session_start();if (!isset($_GET[page])) {show_source(__FILE__);die(); }if (isset($_GET[page]) && $_GET[page] ! index.php) {include(flag.php); }else {header(Location: ?pageflag.php); } <?phpif ($_SESSION[admin]) {$con $_POST[con];$…

可溶性PFA材质三角漏斗耐腐蚀进口聚四氟乙烯漏斗低溶出析出

PFA全名为可溶性聚四氟乙烯、全氟烷氧基树脂&#xff0c;成品外观透明可视&#xff0c;便于观察&#xff0c;有着良好的化学稳定性、耐温性&#xff0c;耐受强酸强碱以及各种有机溶剂&#xff0c;且PFA原料本身较为洁净&#xff0c;金属离子溶出析出少&#xff0c;经过清洗后可…

OpenCV——SUSAN边缘检测

目录 一、SUSAN算法二、代码实现三、结果展示 OpenCV——SUSAN边缘检测由CSDN点云侠原创&#xff0c;爬虫自重。如果你不是在点云侠的博客中看到该文章&#xff0c;那么此处便是不要脸的爬虫。 一、SUSAN算法 Susan边缘检测是一种经典的边缘检测算&#xff0c;它由Susan Smith…

1044: 顺序栈基本操作的实现

解法&#xff1a; #include<iostream> #include<stack> using namespace std; int main() {int n, a, k;stack<int> sk;cin >> n;while (n--) {cin >> a;sk.push(a);}cin >> k;while (k--) {sk.pop();}if (!sk.empty()) {cout << s…

C语言 | 自定义类型:struct结构体(详解)

目录&#xff1a; --前言 1. 结构体类型的定义与基础结构 2. 结构体的使用 3. typedef相关 4. 结构体的自引用 5. 结构体内存对齐 6. 结构体传参 7. 结构体实现位段 --前言&#xff1a; c语言中内置类型&#xff0c;也有自定义的类型。 例如&#xff1a;内置类型 in…

用html写文本变形动画

<!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><title>文本变形动画</title><link rel"stylesheet" href"./style.css"> </head> <body> <!-- 两个文本部分…

给你的Qt软件加个授权

写在前面 环境&#xff1a; Win11 64位 VS2019 Qt5.15.2 核心思路&#xff1a; 将授权相关信息加密保存到License.txt中&#xff0c;软件运行时获取并解密授权信息&#xff0c;判断是否在限制期限内即可。 加解密部分使用第三方openssl库进行&#xff0c;因此需要手动在…

设计模式:时序图

设计模式&#xff1a;时序图 设计模式&#xff1a;时序图时序图元素&#xff08;Sequence Diagram Elements&#xff09;角色&#xff08;Actor&#xff09;对象&#xff08;Object&#xff09;生命线&#xff08;Lifeline&#xff09;控制焦点&#xff08;Focus of Control&am…