谷粒商城-消息队列

商城业务-消息队列-MQ简介

消息队列应用场景一：异步处理

第一种模式我们必须等各个操作的做完才能返回响应，例如：发送邮件、发送短信能不能收到其实并不是侧重点，因此。可以启动两个线程来执行，也就是第二种模式，在此基础上还可以进行优化就是使用消息中间件，将注册消息存入消息队列中让邮件服务、短信服务慢慢去执行从而提升性能。

消息队列应用场景二：应用解耦

例如当我们下订单需要去调用库存系统的接口，但是库存系统的接口经常需要升级，从而导致需要去修改订单系统的源代码，因此，我们可以将订单信息写入消息队列中不管库存系统如何升级，只需要订阅去执行即可从而达到解耦的作用。

消息队列应用场景三：流量控制(流量消锋)

例如秒杀系统，当百万级别的请求向后台发送后台是会宕机的，因此，将请求消息写入消息对了中由后台慢慢的去处理，提高系统的高可用性。 

商城业务-消息队列-RabbitMQ简介

消息代理：替我们接收、发送消息的服务器

目的地：消息发送到哪里

 点对点模式：有很多的消息的接收者，但消息的接受者只能有一个，谁能拿到消息需要靠抢

RabbitMQ是基于AMQP协议实现的并且兼容JMS，ActiveMQ是基于JMS实现的。JMS和AMQP的区别在于：JMS面向纯java平台不不支持跨平台而AMQP是可以跨平台，假如后台服务有用PHP编写则可以兼容。

JMS和AMQP的简单对比 ：

①.AMQP的消息模型中direct exchange是类比JMS中P2P(Queue),AMQP的其它四种消息模型则是类比于JMS的Topic

②.JMS支持的各种消息类型，AMQP只支持byte[]但也无妨最后都可以json序列化后传输

Spring对JMS、AMQP都是支持的并且提供了自动配置和常用注解 

商城业务-消息队列-RabbitMQ工作流程

RabbitMQ的工作流程： 首先，生成者客户端会向消息中间件发送Message，Message由消息头和消息体组成，消息头中有一个route-key属性用于标识存储的队列位置，消息中间件接收到消息之后会由相应的交换机将消息存储到指定的消息队列中，交换机和队列具有绑定关系，无论生成者还是消费者客户端想用发送或者接收消息需要使用connnection去创建一个长连接，长连接类似于高速公里，信道类似于攻速公路中的每个车道。RabbitMQ还有一个虚拟主机即类似于Docker中的容器彼此互不干扰，不需要创建多个RabbitMQ只需要创建多个虚拟机即可实现向java后台、PHP后台发送消息。

长连接的好处是当客户端宕机之后，RabbitMQ将不会向消费者客户端发送消息而是将消息持久化保证消息不会丢失。

商城业务-消息队列-RabbitMQ安装

RabbitMQ的学习传送门：Networking and RabbitMQ — RabbitMQ

docker安装RabbitMQ命令：

docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management

4369, 25672 (Erlang发现&集群端口)
5672, 5671 (AMQP端口)
15672 (web管理后台端口)
61613, 61614 (STOMP协议端口)
1883, 8883 (MQTT协议端口)

 自启动：

docker update rabbitmq --restart=always

登录地址： ip:15672

首次登录的账号密码都是：guest 

OverView介绍：

 查看对应的协议和端口号

RabbitMQ配置文件的迁移，从老版本的RabbitMQ中下载配置文件 

上传至新版本的RabbitMQ配置文件 

Connections介绍：

 Channels介绍： 

 Exchanges介绍：

添加新的交换机

队列介绍： 

Admin介绍： 

虚拟主机介绍： 

查看自己创建的虚拟主机： 

进入后可配置权限

删除虚拟主机 

设置最大连接数 

显示集群消息

商城业务-消息队列-Exchange类型

创建一个交换机 

创建一个队列

将交换机与队列进行绑定

 Unbind:可以解除绑定

商城业务-消息队列-Direct-Exchange

直接交换机：精确匹配路由键

根据这张图创建所有要用的交换机、队列、以及绑定关系

依次创建4个队列 

创建直接交换机 

绑定关系

发布消息

查看消息

Nack：表示收到了消息不告诉服务器，消息队列中的消息数量不会减少   ACK：告诉服务器收到了消息，队列中的消息就没了 

商城业务-消息队列-Fanout-Exchange

扇形交换机是广播的方式，与其绑定的队列，无论是否有路由键都会收到消息

创建扇形交换机

绑定队列 

发布消息

商城业务-消息队列-Topic-Exchange

主题交换机用于模糊匹配，#匹配0个或多个单词，*匹配一个单词

添加交换机

绑定队列

 发布消息

商城业务-消息队列-SpringBoot整合RabbitMQ

使用RabbitMQ的步骤：

1.在订单服务中导入amqp的启动器，自动配置了RabbitAutoConfiguration配置类，为容器添加了CachingConnectionFactory、RabbitTemplate、AmqpAdmin、RabbitMessagingTemplate等工具类。

<dependency>
   <groupId>org.springframework.bootgroupId>
   <artifactId>spring-boot-starter-amqpartifactId>
dependency>

 2.配置

配置文件前缀如下图所示：

 

3. 启动RabbitMQ

商城业务-消息队列-AmqpAdmin使用

测试

1.使用AmqpAdmin创建交换机

交换机的类型如下图所示

2.创建队列 

3.绑定 

商城业务-消息队列-RabbitTemplate使用

1.使用RabbitTemplate工具类发送String类型消息

2. 使用RabbitTemplate工具类发送java对象

前提条件：java对象实现了Serializable接口

3. 使用RabbitTemplate工具类发送json数据 

前提条件：给容器中注入json转化器

import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class MyRabbitConfig {
 
    @Bean
    public AbstractMessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

商城业务-消息队列-RabbitListener&RabbitHandler接收消息

@RabbitListener使用前提：必须有@EnableRabbit并且标注方法的类必须在组件中

@RabbitListener标注类上监听多个队列

@RabbitHandler标注在方法上用于接受不同类型的消息对象

 
接收到消息...内容：
(Body:'{"id":1,"name":"哈哈哈哈","sort":null,"status":null,"createTime":1658160773440}' 
MessageProperties
 [headers={__TypeId__=com.atguigu.gulimall.order.entity.OrderReturnReasonEntity}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=hello-java-directExchange, receivedRoutingKey=helloQueue, deliveryTag=1, consumerTag=amq.ctag-2bgjjZCJFbf4UPsrqraxQA, consumerQueue=hello-java-queue])
==>类型：class org.springframework.amqp.core.Message

 msg：①消息体+消息头 ②class org.springframework.amqp.core.Message类型的对象

方法的参数类型：

1.class org.springframework.amqp.core.Message类型的对象

2.T<发送消息的类型> ：假如发送消息的类型为 OrderReturnReasonEntity 则接受的消息类型也可以为 OrderReturnReasonEntity

接收到消息...内容：
OrderReturnReasonEntity(id=1, name=哈哈哈哈, sort=null, 
status=null, createTime=Tue Jul 19 00:22:07 CST 2022)

 3.Channel channel：当前传输数据的通道

模拟多个客户端监听Queue。只要收到消息，队列就删除消息，而且只能有一个客户端收到此消息的场景： 

1.订单服务启动多个，同一个消息，只能有一个客户端收到

模拟多个订单服务 

模拟发送多个消息

结果查看，说明：同一个消息，只能有一个客户端收到

2.  只有当一个消息完全处理完，方法运行结束，客户端才可以接收下一个消息

结果查看

@RabbitListener标注类上监听多个队列

@RabbitHandler标注在方法上用于接受不同类型的消息对象

模拟向队列发送不同消息对象

@RabbitHandler标注在方法上，重载区分不同的消息

查看结果 

商城业务-消息队列-可靠投递-发送端确认

 

ComfirmCallBack：当生成者发送消息给broker，broker接收时触发的回调函数

开启ComfirmCallBack回调函数逇步骤：

①开启

#开启broker接收消息的ConfirmCallback回调函数
spring.rabbitmq.publisher-confirm-type-correlated=correlated

 ②编写回调函数

发送消息时还可以设置消息的唯一id

查看broker接收消息之后ConfirmCallback回调函数的执行结果：

ReturnCallBack：当交换机由于某些原因未将消息传送到指定的队列时，触发的回调函数

开启ReturnCallBack回调函数的步骤：

①开启

#开启交换机传递消息给队列失败的ReturnCallback回调函数
spring.rabbitmq.publisher-returns=true
#只要抵达队列，以异步发送优先回调我们这个ReturnCallback
spring.rabbitmq.template.mandaray=true

②编写回调函数

打印结果：

Message：(Body:'{"id":1,"name":"hahah0","sort":null,"status":null,"createTime":1658325488429}' 
MessageProperties [headers={spring_returned_message_correlation=20d529cd-7cd0-46da-a4fe-15f6626d9de4, __TypeId__=com.atguigu.gulimall.order.entity.OrderReturnReasonEntity},
contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]) , 
replyCode : 312 , 
replyText : NO_ROUTE , 
exchangeName : hello-java-directExchange , 
routeKey : hello-Queue

商城业务-消息队列-可靠投递-消费端确认

说明： 消费者消费消息默认采用的是自动ACK也就是自动签收，broker通过通道将消息都传递给你之后自动将消息移除队列，这个就是自动ACK。采用自动ACK将会出现一些问题：当消费者接收到许多条消息时，依次处理这些消息但是在此期间宕机了将会导致后续未处理的消息丢失。

解决方案：手动ACK

手动ACK可以看作是签收操作

①配置

#手动ACK设置
spring.rabbitmq.listener.simple.acknowledge-mode=manual

 ②手动ACK回复

消息的拒收/退回 

 /**
  *  deliveryTag
  *  multiple ： 批处理
  *  requeue : 是否重新入队
  */
channel.basicNack(deliveryTag,false,true);

 /**
  *  deliveryTag
  *  requeue : 是否重新入队
  */
hannel.basicReject(deliveryTag,true);

如何签收：
业务成功就应该签收：channel.basicAck(deliveryTag,false);
业务处理失败就应该拒签，让别人处理：channel.basicNack(deliveryTag,false,true);