RabbitMQ实战(三)-高级特性

JavaEdge • 2023-01-03 • 云技术社区 • 289 阅读

0 相关源码

1 你将学到

如何保证消息百分百投递成功
幂等性
如何避免海量订单生成时消息的重复消费
Confirm确认消息、Return返回消息
自定义消费者
消息的ACK与重回队列
限流
TTL
死信队列

2 保证消息的百分百投递成功

2.1 Producer 的可靠性投递

2.1.1 要求

保证消息的成功发出
保证MQ节点的成功接收
发送端收到MQ节点(Broker) 确认应答
完善的消息补偿机制

在实际生产中，很难保障前三点的完全可靠，比如在极端的环境中，生产者发送消息失败了，发送端在接受确认应答时突然发生网络闪断等等情况，很难保障可靠性投递，所以就需要有第四点完善的消息补偿机制。

2.1.2 解决方案

2.1.2.1 方案一:消息信息落库,对消息状态进行打标(常见方案)

将消息持久化到DB并设置状态值,收到Consumer的应答就改变当前记录的状态.

再轮询重新发送没接收到应答的消息,注意这里要设置重试次数.

方案流程图

方案实现流程

比如我下单成功

step1 - 对订单数据入BIZ DB订单库,并对因此生成的业务消息入MSG DB消息库

此处由于采用了两个数据库,需要两次持久化操作,为了保证数据的一致性,有人可能就想着采用分布式事务,但在大厂实践中,基本都是采用补偿机制!

这里一定要保证step1 中消息都存储成功了，没有出现任何异常情况，然后生产端再进行消息发送。如果失败了就进行快速失败机制

对业务数据和消息入库完毕就进入

setp2 - 发送消息到 MQ 服务上，如果一切正常无误消费者监听到该消息，进入

step3 - 生产端有一个Confirm Listener,异步监听Broker回送的响应,从而判断消息是否投递成功

step4 - 如果成功,去数据库查询该消息,并将消息状态更新为1
step5 - 如果出现意外情况，消费者未接收到或者 Listener 接收确认时发生网络闪断，导致生产端的Listener就永远收不到这条消息的confirm应答了，也就是说这条消息的状态就一直为0了，这时候就需要用到我们的分布式定时任务来从 MSG 数据库抓取那些超时了还未被消费的消息，重新发送一遍
此时我们需要设置一个规则，比如说消息在入库时候设置一个临界值timeout，5分钟之后如果还是0的状态那就需要把消息抽取出来。这里我们使用的是分布式定时任务，去定时抓取DB中距离消息创建时间超过5分钟的且状态为0的消息。

step6 - 把抓取出来的消息进行重新投递(Retry Send)，也就是从第二步开始继续往下走

step7 - 当然有些消息可能就是由于一些实际的问题无法路由到Broker，比如routingKey设置不对，对应的队列被误删除了，那么这种消息即使重试多次也仍然无法投递成功，所以需要对重试次数做限制，比如限制3次，如果投递次数大于三次，那么就将消息状态更新为2，表示这个消息最终投递失败,然后通过补偿机制，人工去处理。实际生产中，这种情况还是比较少的，但是你不能没有这个补偿机制，要不然就做不到可靠性了。

思考:该方案在高并发的场景下是否合适

对于第一种方案，我们需要做两次数据库的持久化操作，在高并发场景下显然数据库存在着性能瓶颈.

其实在我们的核心链路中只需要对业务进行入库就可以了，消息就没必要先入库了，我们可以做消息的延迟投递，做二次确认，回调检查。下面然我们看方案二

2.1.2.2 消息延迟投递,两次确认,回调检查(大规模海量数据方案)

大厂经典实现方案

当然这种方案不一定能保障百分百投递成功，但是基本上可以保障大概99.9%的消息是OK的，有些特别极端的情况只能是人工去做补偿了，或者使用定时任务.

主要就是为了减少DB操作

方案流程图

Upstream Service
上游服务,即生产端
Downstream service
下游服务,即消费端
Callback service
回调服务方案实现流程
step1 一定要先将业务消息入库,然后Pro再发出消息,顺序不能错!
step2 在发送消息之后,紧接着Pro再发送一条消息(Second Send Delay Check),即延迟消息投递检查,这里需要设置一个延迟时间,比如5分钟之后进行投递.
step3 Con监听指定的队列,处理收到的消息.
step4 处理完成之后,发送一个confirm消息,也就是回送响应,但是其不是普通的ACK,而是重新生成一条消息,投递到MQ,表示处理成功.
Callback service是一个单独的服务,它扮演MSG DB角色,它通过MQ监听下游服务发送的confirm消息,如果监听到confirm消息,那么就对其持久化到MSG DB.
step6 5分钟之后延迟消息发送到MQ,然后Callback service还是去监听延迟消息所对应的队列，收到Check消息后去检查DB中是否存在消息，如果存在，则不需要做任何处理，如果不存在或者消费失败了，那么Callback service就需要主动发起RPC通信给上游服务，告诉它延迟检查的这条消息我没有找到，你需要重新发送，生产端收到信息后就会重新查询BIZ DB然后将消息发送出去.

设计目的

少做一次DB的存储,在高并发场景下,最关心的不是消息百分百投递成功,而是一定要保证性能，保证能抗得住这么大的并发量。所以能节省数据库的操作就尽量节省，异步地进行补偿.

其实在主流程里面是没有Callback service的，它属于一个补偿的服务，整个核心链路就是生产端入库业务消息，发送消息到MQ，消费端监听队列，消费消息。其他的步骤都是一个补偿机制。

小结

这两种方案都是可行的，需要根据实际业务来进行选择,方案二也是互联网大厂更为经典和主流的解决方案.但是若对性能要求不是那么高,方案一要更简单.

3 幂等性

3.1 什么是幂等性

用户对于同一操作发起的一次请求或者多次请求的结果是一致的

比如数据库的乐观锁,在执行更新操作前,先去数据库查询version,然后执行更新语句,以version作为条件,如果执行更新时有其他人先更新了这张表的数据,那么这个条件就不生效了,也就不会执行操作了,通过这种乐观锁的机制来保障幂等性.

3.2 Con - 幂等性

3.2.1 什么是Con - 幂等性

在业务高峰期最容易产生消息重复消费问题,当Con消费完消息时,在给Pro返回ack时由于网络中断,导致Pro未收到确认信息,该条消息就会重新发送并被Con消费,但实际上该消费者已成功消费了该条消息,这就造成了重复消费.

而Con - 幂等性,即消息不会被多次消费,即使我们收到了很多一样的消息.

3.2.2 主流幂等性实现方案

3.2.2.1 唯一ID+指纹码

核心:利用数据库主键去重

唯一ID:业务表的主键
指纹码：为了区别每次正常操作的码，每次操作时生成指纹码；可以用时间戳+业务编号或者标志位（具体视业务场景而定）
优势
实现简单
弊端
高并发下有数据库写入的性能瓶颈
解决方案
根据ID进行分库分表算法路由

小结

首先我们需要根据消息生成一个全局唯一ID，然后还需要加上一个指纹码。这个指纹码它并不一定是系统去生成的，而是一些外部的规则或者内部的业务规则去拼接，它的目的就是为了保障这次操作是绝对唯一的。

将ID + 指纹码拼接好的值作为数据库主键，就可以进行去重了。即在消费消息前呢，先去数据库查询这条消息的指纹码标识是否存在，没有就执行insert操作，如果有就代表已经被消费了，就不需要管了

3.2.2.2 利用Redis原子性

这里我们使用Redis实现幂等,还需要考虑如下问题

我们是否要进行数据落库,如果落库,那么数据库和缓存如何做到原子性?
如果你想用事务,放弃吧,Redis缓存事务和MySQL事务根本不是同一个事务
如果不落库,那么都存储到缓存中,定时同步的策略如何设置为好?

这里只提用Redis的原子性去解决MQ幂等性重复消费的问题

MQ的幂等性问题根本在于的是生产端未正常接收ACK，可能是网络抖动、网络中断导致

可能的方案

Con在消费开始时将 ID放入到Redis的BitMap中，Pro每次生产数据时，从Redis的BitMap对应位置若不能取出ID，则生产消息发送，否则不进行消息发送。

但是有人可能会说，万一Con，ProRedis命令执行失败了怎么办，虽然又出现重复消费又出现Redis非正常执行命令的可能性极低，但是万一呢？

OK，我们可以在Redis命令执行失败时，将消息落库，每日用定时器，对这种极特殊的消息进行处理。

4 Confirm机制

4.1 什么是Confirm机制

消息的确认
Pro投递消息后,如果Broker收到消息,则会给Pro一个应答
Pro接收应答
用来确定这条消息是否正常地发送到Broker,该法也是消息可靠性投递的核心保障!4.2 Confirm机制流程图

Pro发送消息到Broker,Broker接收到消息后,产生回送响应
Pro中有一个Confirm Listener异步监听响应应答

4.2 实现Confirm机制

在channel上开启确认模式：channel.confirmSelect()
在channel上添加监听：addConfirmListener
监听成功和失败的返回结果，根据具体的结果对消息进行重新发送、或记录日志等后续处理

接下来就让我们根据原理进行实操吧!

Con端
Pro端
启动Con,检查管控台
启动Pro

5 Return机制5.1 什么是Return机制
Return Listener 用于处理一些不可路由的消息
Pro通过指定一个Exchange和Routingkey,把消息送到某一个队列中,然后Con监听队列,进行消费
但如果我们在发送消息时,当前Exchange不存在或者Routingkey路由不到,如果我们要监听这种不可达的消息,就要用到Return Listener5.2 Return机制示意图

5.3 实现Return机制

添加return监听：addReturnListener，生产端去监听这些不可达的消息，做一些后续处理，比如说，记录下消息日志，或者及时去跟踪记录，有可能重新设置一下就好了

在基础的API中的一个关键的配置项:Mandatory

如果为true，则***会接收到路由不可达的消息,然后进行后续处理
如果为false,那么broker端自动删除该消息
Con
Pro
启动Con
启动Pro
由于Pro设置的是一个错误的路由key，所以消费端没有任何打印，而生产端打印了如下内容
如果我们将 Mandatory 属性设置为false，对于不可达的消息会被Broker直接删除，那么Pro就不会进行任何打印了。如果我们的路由key设置为正确的，那么Con能够正确消费，Pro也不会进行任何打印。6 Con - 自定义监听
之前我们都是在代码中编写while循环,通过consumer.nextDelivery方法获取下一条消息,然后进行消费处理
其实我们还有另一种选择,使用自定义的Consumer,它更方便,解耦性更强,也是在实际工作中最常用的使用方式

自定义Con实现只需要继承 DefaultConsumer 类，重写 handleDelivery 方法即可!

6.1 代码实现

自定义Con
Con
Pro
启动Con后,查看管控台
启动Pro,Con接收消息

7 Con - 限流7.1 什么是Con - 限流7.1.1 消息过载场景
假设我们有这样的场景
Rabbitmq服务器有上万条未处理的消息,我们随便打开一个Con - Client,会造成:巨量的消息瞬间全部推送过来,然而我们单个客户端无法同时处理这么多数据!此时很有可能导致服务器崩溃，严重的可能导致线上的故障。
还有一些其他的场景，比如说单个Pro一分钟产生了几百条数据,但是单个Con一分钟可能只能处理60条,这个时候Pro-Con肯定是不平衡的。通常Pro是没办法做限制的。所以Con肯定需要做一些限流措施，否则如果超出最大负载，可能导致Con性能下降，服务器卡顿甚至崩溃等一系列严重后果

因此,我们需要限流

7.1.2 Con - 限流机制

RabbitMQ提供了一种qos (服务质量保证)功能,即在非自动确认消息的前提下,如果一定数目的消息 (通过基于Con或者channel设置Qos的值) 未被确认前,不消费新的消息

不能设置自动签收功能(autoAck = false)
如果消息未被确认,就不会到达Con,目的就是给Pro减压

限流设置API

void BasicQos(uint prefetchSize, ushort prefetchCount, bool global);

prefetchSize: 单条消息的大小限制，Con通常设置为0，表示不做限制
prefetchCount: 一次最多能处理多少条消息
global: 是否将上面设置true应用于channel级别还是取false代表Con级别

prefetchSize和global这两项,RabbitMQ没有实现,暂且不研究
prefetchCount在 autoAck=false 的情况下生效,即在自动应答的情况下该值无效

手工ACK

void basicAck(Integer deliveryTag，boolean multiple)

调用这个方法就会主动回送给Broker一个应答，表示这条消息我处理完了，你可以给我下一条了。参数multiple表示是否批量签收，由于我们是一次处理一条消息，所以设置为false

7.2 实现Con - 限流

自定义Con
Con
Pro
启动Con,查看管控台
启动Pro,开始发送消息,Con接收消息
实现限流,仅仅处理一条消息,其余的都在等待
现在,我们开启ACK应答处理
重新启动Con,发现剩余的2条消息也全都发送并接收了!
我们之前是注释掉手工ACK方法，然后启动消费端和生产端，当时Con只打印一条消息,这是因为我们设置了手工签收,并且设置了一次只处理一条消息,当我们没有回送ACK应答时，Broker端就认为Con还没有处理完这条消息,基于这种限流机制就不会给Con发送新的消息了,所以Con那时只打印了一条消息

8 Con - ACK & 重回队列机制8.1 ACK & NACK当我们设置autoACK=false 时,就可以使用手工ACK方式了,其实手工方式包括了手工ACK与NACK

当我们手工 ACK 时,会发送给Broker一个应答,代表消息处理成功,Broker就可回送响应给Pro.

NACK 则表示消息处理失败,如果设置了重回队列,Broker端就会将没有成功处理的消息重新发送.

使用方式

Con消费时,如果由于业务异常,我们可以手工 NACK 记录日志,然后进行补偿API：void basicNack(long deliveryTag, boolean multiple, boolean requeue)API：void basicAck(long deliveryTag, boolean multiple)
如果由于服务器宕机等严重问题,我们就需要手工 ACK 保障Con消费成功

8.2 重回队列

重回队列是为了对没有处理成功的消息,将消息重新投递给Broker
重回队列,会把消费失败的消息重新添加到队列的尾端,供Con继续消费
一般在实际应用中,都会关闭重回队列,即设置为false

8.3 实现机制

Con,关闭自动签收功能
自定义Con,对第一条消息(序号0)进行NACK，并设置重回队列
Pro 对消息设置序号,以便区分
启动Con,查看管控台
启动Pro,这里第一条消息由于我们调用了NACK，并且设置了重回队列，所以会导致该条消息一直重复发送，消费端就会一直循环消费

9 TTL机制9.1 什么是TTL
TTL(Time To Live),即生存时间
RabbitMQ支持消息的过期时间，在消息发送时可以进行指定
RabbitMQ支持为每个队列设置消息的超时时间，从消息入队列开始计算，只要超过了队列的超时时间配置，那么消息会被自动清除

9.2 管控台演示

新增一个Q
新增一个交换机
绑定
Q中也显示了相关的绑定信息
发送消息
10s后,消息被清除为0

10 死信队列机制10.1 什么是死信队列DLX - 死信队列(dead-letter-exchange)
利用DLX,当消息在一个队列中变成死信 (dead message) 之后,它能被重新publish到另一个Exchange中,这个Exchange就是DLX.10.2 死信队列的产生场景
消息被拒绝(basic.reject / basic.nack),并且requeue = false
消息因TTL过期
队列达到最大长度10.3 死信的处理过程
DLX亦为一个普通的Exchange,它能在任何队列上被指定,实际上就是设置某个队列的属性
当某队列中有死信时,RabbitMQ会自动地将该消息重新发布到设置的Exchange,进而被路由到另一个队列
可以监听这个队列中的消息做相应的处理.该特性可以弥补RabbitMQ 3.0以前支持的immediate参数的功能

10.4 死信队列的配置

设置死信队列的exchange和queue,然后进行绑定
- Exchange:dlx.exchange
- Queue: dlx.queue
- RoutingKey:#
正常声明交换机、队列、绑定，只不过我们需要在队列加上一个参数即可arguments.put(" x-dead-letter-exchange"，"dlx.exchange");这样消息在过期、requeue、队列在达到最大长度时，消息就可以直接路由到死信队列！10.5 实操演示
自定义Con
Pro
Con
启动Con,查看管控台
现在,让我们停止Con,并启动Pro,由于没有Con,TTL为10s的消息将送往死信队列
10s后
实际环境我们还需要对死信队列进行一个监听和处理，当然具体的处理逻辑和业务相关，这里只是简单演示死信队列是否生效。