RabbitMq消息队列

"MQ"消息处理方式，在当今互联网中运用极其之广，当我们看到一些棘手，很消耗服务资源时，我们开发小哥哥，大佬们总会给出一些解决的方案，【RabbitMq】就是其中一种，比如我们在使用app购买东西下单时，会给用户发送消息&订阅时，我们会把订阅，或者发送消息放在队列中，【异步处理】，不影响主流程下单，下单不会因为消息发不出去，而导致下不了单，这是用户不能接受的。所以在当今Mq消息是大受好评的。

“RabbitMq”到底是什么呢？

**rabbitMq官网地址**

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而群集和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

“MQ”到底能干啥呢？

1.1异步处理

1.2应用解耦

1.3流量控制

概述

1. 大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力
2. 消息服务中两个重要概念： 消息代理（message broker）和目的地（destination） 当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。

“消息队各种模式”

2.1简单模式（Hello World）

做最简单的事情，一个生产者对应一个消费者，RabbitMQ相当于一个消息代理，负责将A的消息转发给B

应用场景： 将发送的电子邮件放到消息队列，然后邮件服务在队列中获取邮件并发送给收件人

2.2工作队列模式（Work queues）

一次向许多消费者发送消息，一个生产者发送的消息会被多个消费者获取，也就是将消息将广播到所有的消费者中。

应用场景： 更新商品库存后需要通知多个缓存和多个数据库，这里的结构应该是：

• 一个fanout类型交换机扇出两个个消息队列，分别为缓存消息队列、数据库消息队列
• 一个缓存消息队列对应着多个缓存消费者
• 一个数据库消息队列对应着多个数据库消费者

2.3 路由模式（Routing）

有选择地（Routing key）接收消息，发送消息到交换机并且要指定路由key
，消费者将队列绑定到交换机时需要指定路由key，仅消费指定路由key的消息

应用场景： 如在商品库存中增加了1台iphone12，iphone12促销活动消费者指定routing
key为iphone12，只有此促销活动会接收到消息，其它促销活动不关心也不会消费此routing key的消

2.4 主题模式（Topics）

根据主题（Topics）来接收消息，将路由key和某模式进行匹配，此时队列需要绑定在一个模式上，#匹配一个词或多个词，*只匹配一个词。

应用场景： 同上，iphone促销活动可以接收主题为iphone的消息，如iphone12、iphone13等

2.5远程过程调用（RPC）

如果我们需要在远程计算机上运行功能并等待结果就可以使用RPC，具体流程可以看图。应用场景：需要等待接口返回数据，如订单支付

2.6发布者确认（Publisher /confirm/is）

与发布者进行可靠的发布确认，发布者确认是RabbitMQ扩展，可以实现可靠的发布。在通道上启用发布者确认后，RabbitMQ将异步确认发送者发布的消息，这意味着它们已在服务器端处理。

应用场景： 对于消息可靠性要求较高，比如钱包扣款

java框架的兼容

Spring支持
• spring-jms提供了对JMS的支持
• spring-rabbit提供了对AMQP的支持
• 需要ConnectionFactory的实现来连接消息代理
• 提供JmsTemplate、RabbitTemplate来发送消息
• @JmsListener（JMS）、@RabbitListener（AMQP）注解在方法上监听消息
代理发布的消息
• @EnableJms、@EnableRabbit开启支
Spring Boot自动配置
• JmsAutoConfiguration
• RabbitAutoConfiguration
• 10、市面的MQ产品
• ActiveMQ、RabbitMQ、RocketMQ、Kafka

RabbitMQ概念

Message
消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，
这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可
能需要持久性存储）等。
Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Exchange有4种类型：direct(默认)，fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别
Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直
在队列里面，等待消费者连接到这个队列将其取走。
Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交
换器理解成一个由绑定构成的路由表。
Exchange 和Queue的绑定可以是多对多的关系。
Connection
网络连接，比如一个TCP连接。
Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道
发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都
是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加
密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥
有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时
指定，RabbitMQ 默认的 vhost 是 / 。
Broker
表示消息队列服务器实体

工作流程图

RabbitMQ运行机制

AMQP 中消息的路由过程和 Java 开 发者熟悉的 JMS 存在一些差别， AMQP 中增加了 Exchange 和 Binding
的角色。生产者把消息发布 到 Exchange 上，消息最终到达队列 并被消费者接收，而 Binding 决定交
换器的消息应该发送到那个队列

MQ整合springboot

配置文件（单机版的）

# RabbitMQ配置
spring.rabbitmq.host=192.168.77.130
spring.rabbitmq.port=5672
# 虚拟主机配置
spring.rabbitmq.virtual-host=/
# 开启发送端消息抵达Broker确认
spring.rabbitmq.publisher-/confirm/is=true
# 开启发送端消息抵达Queue确认
spring.rabbitmq.publisher-returns=true
# 只要消息抵达Queue，就会异步发送优先回调returnfirm
spring.rabbitmq.template.mandatory=true
# 手动ack消息，不使用默认的消费端确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual

**测试MQ**
1. AmqpAdmin：管理组件
2. RabbitTemplate：消息发送处理组件
3. @RabbitListener 监听消息的方法可以有三种参数（不分数量，顺序）
• Object content, Message message, Channel 

监听消息

MyRabbitConfig 配置（定制）

@Configuration
public class MyRabbitConfig {

    private RabbitTemplate rabbitTemplate;

    @Primary
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setMessageConverter(messageConverter());
        initRabbitTemplate();
        return rabbitTemplate;
    }

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    
    // @PostConstruct  //MyRabbitConfig对象创建完成以后，执行这个方法
    public void initRabbitTemplate() {

        
        //设置确认回调
        rabbitTemplate.set/confirm/iCallback((correlationData,ack,cause) -> {
            System.out.println("/confirm/i...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]");
        });


        
        rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> {
            System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]" +
                    "==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]");
        });
    }
}    

RabbitMQ消息确认机制-可靠抵达

• 保证消息不丢失，可靠抵达，可以使用事务消息，性能下降250倍，为此引入确认 机制 • publisher /confirm/iCallback
确认模式 • publisher returnCallback 未投递到 queue 退回模式 • consumer ack机制

3.1 可靠抵达-/confirm/iCallback

• spring.rabbitmq.publisher-confirms=true
• 在创建 connectionFactory 的时候设置 Publisher/confirm/is(true) 选项，开启
  confirmcallback 。
• CorrelationData：用来表示当前消息唯一性。
• 消息只要被 broker 接收到就会执行 /confirm/iCallback，如果是 cluster 模式，需要所有
  broker 接收到才会调用 /confirm/iCallback。
• 被 broker 接收到只能表示 message 已经到达服务器，并不能保证消息一定会被投递
  到目标 queue 里。所以需要用到接下来的 returnCallbac

3.2 可靠抵达-ReturnCallback

• spring.rabbitmq.publisher-returns=true
• spring.rabbitmq.template.mandatory=true
• confrim 模式只能保证消息到达 broker，不能保证消息准确投递到目标 queue 里。在有
  些业务场景下，我们需要保证消息一定要投递到目标 queue 里，此时就需要用到
  return 退回模式。
• 这样如果未能投递到目标 queue 里将调用 returnCallback ，可以记录下详细到投递数
  据，定期的巡检或者自动纠错都需要这些数据

3.3 可靠抵达-Ack消息确认机制

•消费者获取到消息，成功处理，可以回复Ack给Broker
•basic.ack用于肯定确认；broker将移除此消息
•basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
•basic.reject用于否定确认；同上，但不能批量
•默认自动ack，消息被消费者收到，就会从broker的queue中移除
•queue无消费者，消息依然会被存储，直到消费者消费
•消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，
 或者成功处理。我们可以开启手动ack模式
•消息处理成功，ack()，接受下一个消息，此消息broker就会移除
•消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
•消息一直没有调用ack/nack方法，broker认为此消息正在被处理，不会投递给别人，此时客户
端断开，消息不会被broker移除，会投递给别人

RabbitMQ延时队列（实现定时任务

场景： 比如未付款订单，超过一定时间后，系统自动取消订单并释放占有物品。 常用解决方案： spring的 schedule
定时任务轮询数据库 缺点： 消耗系统内存、增加了数据库的压力、存在较大的时间误差
解决：rabbitmq的消息TTL和死信Exchange结合

消息的TTL（Time To Live）

• 消息的TTL就是消息的存活时间。
• RabbitMQ可以对队列和消息分别设置TTL。
• 对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的
设置。超过了这个时间，我们认为这个消息就死了，称之为死信。
• 如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队
列中，这个消息死亡的时间有可能不一样（不同的队列设置）。
TTL，因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x- message-ttl属性来设置时间，两者是一样的效果

Dead Letter Exchanges（DLX）

• 一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，
一个路由可以对应很多队列。（什么是死信）
• 一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。也就是说不
会被再次放在队列里，被其他消费者使用。（basic.reject/ basic.nack）requeue=false
• 上面的消息的TTL到了，消息过期了。
• 队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上
• Dead Letter Exchange其实就是一种普通的exchange，和创建其他
exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有
消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。
• 我们既可以控制消息在一段时间后变成死信，又可以控制变成死信的消息
被路由到某一个指定的交换机，结合二者，其实就可以实现一个延时队列
• 手动ack&异常消息统一放在一个队列处理建议的两种方式
• catch异常后，手动发送到指定队列，然后使用channel给rabbitmq确认消息已消费
• 给Queue绑定死信队列，使用nack（requque为false）确认消息消费失败

延时队列实现-1

延时队列实现 业务

如何保证消息可靠性-消息丢失

• 1、消息丢失
• 消息发送出去，由于网络问题没有抵达服务器
• 做好容错方法（try-catch），发送消息可能会网络失败，失败后要有重试机
制，可记录到数据库，采用定期扫描重发的方式
• 做好日志记录，每个消息状态是否都被服务器收到都应该记录
• 做好定期重发，如果消息没有发送成功，定期去数据库扫描未成功的消息进
行重发
• 消息抵达Broker，Broker要将消息写入磁盘（持久化）才算成功。此时Broker尚
未持久化完成，宕机。
• publisher也必须加入确认回调机制，确认成功的消息，修改数据库消息状态。
• 自动ACK的状态下。消费者收到消息，但没来得及消息然后宕机
• 一定开启手动ACK，消费成功才移除，失败或者没来得及处理就noAck并重
新入队

如何保证消息可靠性-消息重复

消息重复
• 消息消费成功，事务已经提交，ack时，机器宕机。导致没有ack成功，Broker的消息
重新由unack变为ready，并发送给其他消费者
• 消息消费失败，由于重试机制，自动又将消息发送出去
• 成功消费，ack时宕机，消息由unack变为ready，Broker又重新发送
• 消费者的业务消费接口应该设计为幂等性的。比如扣库存有
工作单的状态标志
• 使用防重表（redis/mysql），发送消息每一个都有业务的唯
一标识，处理过就不用处理
• rabbitMQ的每一个消息都有redelivered字段，可以获取是否
是被重新投递过来的，而不是第一次投递过来的

如何保证消息可靠性-消息积压

消息积压
• 消费者宕机积压
• 消费者消费能力不足积压
• 发送者发送流量太大
• 上线更多的消费者，进行正常消费
• 上线专门的队列消费服务，将消息先批量取出来，记录数据库，离线慢慢处理

ActiveMQ、RabbitMQ、RocketMQ、Kafka（消息中间件对比）

mq使用场景图

很多来源于【bibi】&尚硅谷 ，有兴趣小伙伴可以关注，创新不多，自当留用于笔记
@lcc