MQ消息队列的基础使用（以RabbitMQ为例）

一、同步调用与异步调用

1、同步调用

在这里插入图片描述
同步调用优点：

时效性强，可以立即得到结果

同步调用存在的问题

（1）耦合度高：每次加入新的需求，都要修改原来的代码

（2）性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和

（3）资源浪费：调用链中的每个服务在等待响应过长中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源

（4）级联失败：如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

2、异步调用

在这里插入图片描述
优点：

耦合度低
吞吐量提高
故障隔离
流量削峰

缺点：

依赖于Broker的可靠性、安全性、吞吐能力
架构复杂，业务没有明显的流程线，不好追踪管理

二、初识MQ

MQ（MessageQueue），中文是消息队列，字面意思就是存放消息的队列。也就是事件驱动架构中的Broker。
在这里插入图片描述

三、RabbitMQ

docker安装RabbitMQ步骤

1、拉取镜像

docker pull rabbitmq

2、安装RabbitMQ，执行命令

docker run -d--restart=always \-p15672:15672 -p5672:5672 \-eRABBITMQ_DEFAULT_USER=admin \-eRABBITMQ_DEFAULT_PASS=Miss5016 \--hostname rabbitmq \--name rabbitmq \
        rabbitmq:latest

3、开放防火墙和安全组对应的端口

# 开放端口防火墙
firewall-cmd --permanent--zone=public --add-port=15672/tcp
firewall-cmd --permanent--zone=public --add-port=5672/tcp
# 重启防火墙
systemctl restart firewalld.service

4、启用RabbitMQ的web端管理界面

# 进入容器内部dockerexec-it my-rabbitmq /bin/bash
# 允许使用web端管理
rabbitmq-plugins enable rabbitmq_management

5、通过浏览器访问，ip+端口号
在这里插入图片描述
6、输入docker run命令时的账号密码即可进入访问

7、问题

通道被禁用 Stats in management UI are disabled on this node
在这里插入图片描述
解决方法：

（1）进入容器内部

dockerexec-it my-rabbitmq /bin/bash

（2）切换到RabbitMQ的配置目录下

cd /etc/rabbitmq/conf.d/

（3）修改配置

echo management_agent.disable_metrics_collector =false> management_agent.disable_metrics_collector.conf

（4）退出并重启docker容器

# 从容器内部退出exit# 重启容器docker restart my-rabbitmq

（5）重新访问通道
在这里插入图片描述
RabbitMQ的结构和概念

在这里插入图片描述

四、RabbitMQ的消息模型

官网文档：https://www.rabbitmq.com/tutorials

1、基本消息队列（BasicQueue）

在这里插入图片描述
总共分为三个角色：

在这里插入图片描述

publisher：消息发布者，将消息发送到队列queue
queue：消息队列，负责接受消息并缓存消息
consumer：订阅队列，处理队列中的消息

示例，通过Springboot的测试类实现

（1）导入依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version></dependency>

（2）生产者代码

@TestvoidcontextLoads(){try{//1、建立连接ConnectionFactory factory =newConnectionFactory();//2、设置连接参数
            factory.setHost("ip地址");//主机名
            factory.setPort(5672);//连接端口
            factory.setVirtualHost("/");//虚拟主机
            factory.setUsername("用户名");//用户名
            factory.setPassword("密码");//密码//3、建立连接Connection connection = factory.newConnection();//4、创建通道Channel channel = connection.createChannel();//5、创建队列并声明队列String queueName ="simple.queue";//队列名称
            channel.queueDeclare(queueName,false,false,false,null);//6、发送消息String message ="hello rabbitMQ";
            channel.basicPublish("", queueName,null, message.getBytes());System.out.println("消息发送完毕。。。");//关闭管道和连接
            channel.close();
            connection.close();}catch(Exception e){System.out.println("生产者发送消息失败 ----- > "+ e.getMessage());}}

（3）消费者代码

@TestvoidcontextLoads(){try{//1、建立连接ConnectionFactory factory =newConnectionFactory();//2、设置连接参数
            factory.setHost("ip地址");//主机名
            factory.setPort(5672);//连接端口
            factory.setVirtualHost("/");//虚拟主机
            factory.setUsername("用户名");//用户名
            factory.setPassword("密码");//密码//3、建立连接Connection connection = factory.newConnection();//4、创建通道Channel channel = connection.createChannel();//5、创建队列并声明队列String queueName ="simple.queue";//队列名称
            channel.queueDeclare(queueName,false,false,false,null);//6、订阅消息
            channel.basicConsume(queueName,true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{//将字节流内容转为消息本体String message =newString(body);System.out.println("消费者接收到消息 --------- > "+ message);}});System.out.println("等待消息中。。。。");//关闭管道和连接
            channel.close();
            connection.close();}catch(Exception e){System.out.println("生产者发送消息失败 ----- > "+ e.getMessage());}}

基本消息队列的消息发送流程：

建立connection
创建channel
利用channel声明队列
利用channel向队列发送消息

基本消息队列的消息接收流程

建立connection
创建channel
利用channel声明队列
定义consumer的消费行为handleDelivery()
利用channel将消费者与队列绑定

2、工作消息队列（WorkQueue）

在这里插入图片描述

3、发布订阅（Publish、Subscribe）模式

根据交换机的类型，可分为不同的三种

（1）广播（Fanout Exchange）
在这里插入图片描述
（2）路由（Direct Exchange）

（3）主题（Topic Exchange）

五、SpringAMQP

1、介绍

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

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

2、示例

简单消息队列

发送消息：

（1）在父工程中引入依赖，确保每个子工程都能使用

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

（2）编写配置类

spring:rabbitmq:host:# 消息队列ipport:# 端口virtual-host:# 虚拟地址username:# 用户名password:# 密码

（3）测试类

@SpringBootTestclassOrderServiceApplicationTests{@AutowiredprivateRabbitTemplate rabbitTemplate;/**
     * 发送消息
     */@TestvoidsendMessageMQ(){//队列名称String queueName ="simple.queue";//消息String message ="你好，rabbitmq";//发送消息
        rabbitTemplate.convertAndSend(queueName, message);}}

接收消息：

（1）引入依赖

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

（2）编写配置类

spring:rabbitmq:host:# 消息队列ipport:# 端口virtual-host:# 虚拟地址username:# 用户名password:# 密码

（3）接收并处理消息

@Component//监听消息的监听器publicclassSpringRabbitListener{//需要监听哪个消息队列中的消息@RabbitListener(queues ="simple.queue")publicvoidlistenerSimpleQueue(String message){//处理消息System.out.println("监听到 simple.queue 队列中的消息 --------- > "+ message);}}

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

Work Queue 工作队列

提高消息处理的速度，避免消息堆积

示例：

生产者，生产100条消息

@TestvoidsendMessageWorkQueue()throwsInterruptedException{String queueName ="simple.queue";String message ="你好，rabbitmq__";for(int i =1; i <=100; i++){
            rabbitTemplate.convertAndSend(queueName, message+i);Thread.sleep(20);}}

两个消费者，共同消费，设置不同的处理速度

@ComponentpublicclassSpringRabbitListener{@RabbitListener(queues ="simple.queue")publicvoidlistenerWork1Queue(String message)throwsInterruptedException{System.out.println("消费者1消费了消息 --------- > "+ message +" "+LocalDateTime.now());Thread.sleep(20);}@RabbitListener(queues ="simple.queue")publicvoidlistenerWork2Queue(String message)throwsInterruptedException{System.err.println("消费者2......消费了消息 --------- > "+ message +" "+LocalDateTime.now());Thread.sleep(200);}}

结果：即使设置了不同的处理速度，但是两个消费者仍然处理了相同的消息数量，并没有因为处理得快的消费者会处理更多消息

原因：因为rabbitmq中的一个消息预取机制造成的

消息预取：当有大量消息到达消息队列时，队列会进行消息投递，消费者的通道（channel）会预先把消息取出，就会平均分配队列中的消息，并不会因为消费者的处理能力进行分配

控制方法：

修改配置文件，设置值，可以控制预取消息的上限

spring:rabbitmq:host:# 消息队列ipport:# 端口virtual-host:# 虚拟地址username:# 用户名password:# 密码listener:simple:prefetch:1# 每次只能取1条消息，处理完后再取

发布（Publish）、订阅（Subscribe）

发布订阅模式允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。

在这里插入图片描述

注意：exchange（交换机）负责消息路由，而不是存储，路由失败则消息丢失

发布订阅-Fanout Exchange（广播）

Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue（队列）

在这里插入图片描述

示例：

（1）在消费者中声明，将交换机与队列进行绑定

声明两个队列和一个交换机，进行绑定

@ConfigurationpublicclassFanoutConfig{/**
     * 声明交换机
     * @return
     */@BeanpublicFanoutExchangefanoutExchange(){//返回一个声明的交换机returnnewFanoutExchange("fanout.exchange");}/**
     * 声明第一个队列
     * @return
     */@BeanpublicQueuefanoutQueue1(){returnnewQueue("fanout.queue1");}/**
     * 声明第二个队列
     * @return
     */@BeanpublicQueuefanoutQueue2(){returnnewQueue("fanout.queue2");}/**
     * 绑定队列到交换机
     * @param fanoutQueue1
     * @param fanoutExchange
     * @return
     */@BeanpublicBindingbindingFanoutQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){//将队列1绑定到交换机returnBindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/**
     * 绑定队列到交换机
     * @param fanoutQueue2
     * @param fanoutExchange
     * @return
     */@BeanpublicBindingbindingFanoutQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){//将队列2绑定到交换机returnBindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}}

（2）处理消息

@ComponentpublicclassSpringRabbitListener{@RabbitListener(queues ="fanout.queue1")publicvoidlistenerFanoutQueue1(String message){System.out.println("监听到 fanout.queue1 队列中的消息 --------- > "+ message);}@RabbitListener(queues ="fanout.queue2")publicvoidlistenerFanoutQueue2(String message){System.out.println("监听到 fanout.queue2 队列中的消息 --------- > "+ message);}}

（3）发送消息

/**
 * 发送消息到交换机
 */@TestvoidtestSendFanoutExchange(){String exChangeName ="fanout.exchange";String message ="交换机转发的消息";
    rabbitTemplate.convertAndSend(exChangeName,"", message);}

这样就实现了每个队列都能消费消息，而不是消息一旦被其中一个消费者消费后，就会移除的问题

发布订阅-DirectExchange（路由模式）

Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。

每一个Queue都会与Exchange设置一个BindingKey
发布者发送消息时，指定消息的RoutingKey
Exchange将消息路由到BindingKey与消息RoutingKey一致的队列示例：

（1）声明队列，交换机，通过注解的方式

@ComponentpublicclassSpringDirectListener{/**
     * 声明第一个消费者
     * @param message
     */@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="direct.queue1"),
            exchange =@Exchange(name ="direct.exchange", type =ExchangeTypes.DIRECT),
            key ={"red","blue"}))publicvoidlistenerDirect1Queue(String message){System.out.println("direct消费者1监听到的消息 ---------- > "+ message);}/**
     * 声明第二个消费者
     * @param message
     */@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="direct.queue2"),
            exchange =@Exchange(name ="direct.exchange", type =ExchangeTypes.DIRECT),
            key ={"red","yellow"}))publicvoidlistenerDirect2Queue(String message){System.out.println("direct消费者2监听到的消息 ---------- > "+ message);}}

（2）发消息

/**
 * 发送消息到交换机
 */@TestvoidtestSendDirectExchange(){String exChangeName ="direct.exchange";String message ="direct交换机转发的消息";String routingKey ="red";//发送给指定routingKey的队列
    rabbitTemplate.convertAndSend(exChangeName, routingKey, message);}

发布订阅-TopicExchange（主题模式）

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且用 . 分割
在这里插入图片描述

示例：

（1）声明交换机，队列

@ComponentpublicclassSpringTopicListener{@RabbitListener(bindings =@QueueBinding(
            value =@Queue("topic.queue1"),
            exchange =@Exchange(value ="topic.exchange", type =ExchangeTypes.TOPIC),
            key ={"china.#"}))publicvoidlistenerTopic1Queue(String message){System.out.println("Topic消费者1监听到的消息 ---------- > "+ message);}@RabbitListener(bindings =@QueueBinding(
            value =@Queue("topic.queue2"),
            exchange =@Exchange(value ="topic.exchange", type =ExchangeTypes.TOPIC),
            key ={"#.news"}))publicvoidlistenerTopic2Queue(String message){System.out.println("Topic消费者2监听到的消息 ---------- > "+ message);}}

（2）测试消息发送

/**
 * 发送消息到交换机
 */@TestvoidtestSendTopicExchange(){String exChangeName ="topic.exchange";String message ="topic交换机转发的消息";String routingKey ="11.news";
    rabbitTemplate.convertAndSend(exChangeName, routingKey, message);}

六、消息转换器

测试：

（1）新增一个队列（queue）

/**
 * 声明一个队列
 * @return
 */@BeanpublicQueueobjectQueue(){returnnewQueue("object.queue");}

（2）测试发送对象类型数据

/**
 * 发送消息
 */@TestvoidsendObjectMessageMQ(){String queueName ="object.queue";Map<String,Object> map =newHashMap<>();
    map.put("username","张三");
    map.put("age",12);
    rabbitTemplate.convertAndSend(queueName, map);}

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

原因：

rabbitmq只支持字节，而spring允许发object对象，说明spring会将对象做序列化，用的序列化方式是jdk的序列化方式x-java-serialized-object，这种序列化缺点是性能差，安全性有问题，可能存在注入风险，数据长度太长了

Spring的对象处理事由org.springframework.amqp.support.converter.MessageConverter来处理的，而默认实现的SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改需要定义一个MessageConverter类型的Bean即可。

步骤：

（1）导入依赖

<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml --><dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.17.1</version></dependency>

（2）声明一个bean，将消息转换的bean注入spring容器中

/**
 * 声明消息转换类型
 * @return
 */@BeanpublicMessageConvertermessageConverter(){returnnewJackson2JsonMessageConverter();}

（3）将消息清除，重新测试
在这里插入图片描述
消息接收：

（1）引入依赖

<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml --><dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.17.1</version></dependency>

（2）注入bean

/**
 * 声明消息转换类型
 * @return
 */@BeanpublicMessageConvertermessageConverter(){returnnewJackson2JsonMessageConverter();}

（3）消息接收处理

@RabbitListener(queues ="object.queue")publicvoidlistenerObjectQueue(Map<String,Object> map){System.out.println("监听到 object.queue 队列中的消息 --------- > "+ map);}

标签： rabbitmq 分布式

本文转载自: https://blog.csdn.net/weixin_47350057/article/details/139749941
版权归原作者 时光荣不掉的冰花窗 所有，如有侵权，请联系我们删除。

MQ消息队列的基础使用（以RabbitMQ为例）

一、同步调用与异步调用

1、同步调用

2、异步调用

二、初识MQ

三、RabbitMQ

四、RabbitMQ的消息模型

1、基本消息队列（BasicQueue）

2、工作消息队列（WorkQueue）

3、发布订阅（Publish、Subscribe）模式

五、SpringAMQP

1、介绍

2、示例

简单消息队列

Work Queue 工作队列

发布（Publish）、订阅（Subscribe）

发布订阅-Fanout Exchange（广播）

发布订阅-DirectExchange（路由模式）

发布订阅-TopicExchange（主题模式）

六、消息转换器

发表评论

“MQ消息队列的基础使用（以RabbitMQ为例）”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航