php raabitmq中间件_消息中间件——RabbitMQ(七)高级特性全在这里!(上)
发布日期:2021-10-31 12:51:53 浏览次数:7 分类:技术文章

本文共 4509 字,大约阅读时间需要 15 分钟。

求关注

高级特性全在这里!(上)

前言

前面我们介绍了RabbitMQ的安装、各大消息中间件的对比、AMQP核心概念、管控台的使用、快速入门RabbitMQ。本章将介绍RabbitMQ的高级特性。分两篇(上/下)进行介绍。消息如何保障100%的投递成功?

幂等性概念详解

在海量订单产生的业务高峰期,如何避免消息的重复消费的问题?

Confirm确认消息、Return返回消息

1 消息如何保障100%的投递成功?

1.1 什么是生产端的可靠性投递?保障消息的成功发出

保障MQ节点的成功接收

发送端收到MQ节点(Broker)确认应答

完善的消息进行补偿机制

前三步不一定能保障消息能够100%投递成功。因此要加上第四步

BAT/TMD 互联网大厂的解决方案:

- 消息落库,对消息状态进行打标

在发送消息的时候,需要将消息持久化到数据库中,并给这个消息设置一个状态(未发送、发送中、到达)。当消息状态发生了变化,需要对消息做一个变更。针对没有到达的消息做一个轮训操作,重新发送。对轮训次数也需要做一个限制3-5次。确保消息能够成功的发送.消息的延迟投递,做二次确认,回调检查

具体采用哪种方案,还需要根据业务与消息的并发量而定。

1.2 第一种方案:

生产端-可靠性投递

第一种方案

图解:

蓝色部分表示:生产者负责发送消息发送至Broker端

Biz DB:订单数据库 MSG DB: 消息数据

面对小规模的应用可以采用加事务的方式,保证事务的一致性。但在大厂中面对高并发,并没有加事务,事务的性能拼接非常严重,而是做补偿。

比如:如下发一条订单消息。

step1:存储订单消息(创建订单),业务数据入库,消息也入库。缺点:需要持久化两次。(status:0)

step2:在step1成功的前提下,发送消息

step3:Broker收到消息后,confirm给我们的生产端。Confirm Listener异步监听Broker回送的消息。

step4:抓取出指定的消息,更新(status=1),表示消息已经投递成功。

step5:分布式定时任务获取消息状态,如果等于0则抓取数据出来。

step6:重新发送消息

step7:重试限制设置3次。如果消息重试了3次还是失败,那么(status=2),认为这个消息就是失败的。

查询这些消息为什么失败,可能需要人工去查询。

假设step2执行成功,step3由于网络闪断。那么confirm将永远收不到消息,那么我们需要设定一个规则:

例如:在消息入库的时候,设置一个临界值 timeout=5min,当超过5min之后,就将这条数据抓取出来。

或者写一个定时任务每隔5分钟就将status=0的消息抓取出来。可能存在小问题:消息发送出去,定时任务又正好刚执行,Confirm还未收到,定时任务就会执行,会导致消息执行两次。

更精细化操作:消息超时容忍限制。confirm在2-3分钟内未收到消息,则重新发送。保障MQ我们思考如果第一种可靠性投递,在高并发的场景下是否合适?

第一种方案对数据有两次入库,一次业务数据入库,一次消息入库。这样对数据的入库是一个瓶颈。

其实我们只需要对业务进行入库。

消息的延迟投递,做二次确认,回调检查

这种方式并不一定能保证100%成功,但是也能保证99.99%的消息成功。如果遇到特别极端的情况,那么就只能需要人工去补偿,或者定时任务去做。

第二种方式主要是为了减少对数据库的操作。

看下第二种方式:

第二种方案

图解:

Upstream service:生产端

DownStream service:消费端

Callback service:回调服务

step1:业务消息入库成功后,第一次消息发送。

step2:同样在消息入库成功后,发送第二次消息,这两条消息是同时发送的。第二条消息是延迟检查,可以设置2min、5min 延迟发送。

step3:消费端监听指定队列。

step4:消费端处理完消息后,内部生成新的消息send confirm。投递到MQ Broker。

step5: Callback Service 回调服务监听MQ Broker,如果收到Downstream service发送的消息,则可以确定消息发送成功,执行消息存储到MSG DB。

step6:Check Detail检查监听step2延迟投递的消息。此时两个监听的队列不是同一个,5分钟后,Callback service收到消息,检查MSG DB。如果发现之前的消息已经投递成功,则不需要做其他事情。如果检查发现失败,则Callback 进行补偿,主动发送RPC 通信。通知上游生产端重新发送消息。这样做的目的:少做了一次DB存储。关注点并不是百分百的投递成功,而是性能。

2. 幂等性概念

2.1 幂等性是什么?

幂等(idempotent、idempotence)是一个数学与计算机学概念,常见于抽象代数中,即f(f(x)) = f(x)。简单的来说就是一个操作多次执行产生的结果与一次执行产生的结果一致。我们可以借鉴数据库的乐观锁机制:

比如我们执行一条更新库存的SQL语句:

UPDATE T_REPS SET COUNT = COUNT - 1,VERSION = VERSION + 1 WHERE VERSION = 1

利用加版本号Version的方式来保证幂等性。

推荐文章:面试必备的数据库悲观锁与乐观锁

2.2 消费端-幂等性保障

在海量订单产生的业务高峰期,如何避免消息的重复消费问题?

在高并发的情况下,会有大量的消息到达MQ,消费端需要监听大量的消息。这样的情况下,难免会出现消息的重复投递,网络闪断等等。如果不去做幂等,则会出现消息的重复消费。

-消费端实现幂等性,就意味着,我们的消息永远不会被消费多次,即使我们收到了多条一样的消息,也只会执行一次。

看下互联网大厂主流的幂等性操作:

-唯一ID+指纹吗机制,利用数据库主键去重。

-利用Redis的原子性实现

-其他的技术实现幂等性

2.2.1 唯一ID+指纹码机制唯一ID + 指纹吗机制,利用数据库主键去重。

保证唯一性

SELECT COUNT(1) FROM T_ORDER WHERE ID = 唯一ID + 指纹码

如果查询没有,则添加。有则不需要做任何操作,消费端不需要消费消息。

好处:实现简单

坏处:高并发下有数据库写入的性能瓶颈

解决方案:跟进ID进行分库分表进行算法路由

分摊流量压力。

2.2.2 Redis 原子特性实现

最简单使用Redis的自增。使用Redis进行幂等,需要考虑的问题。

第一:我们是否需要数据落库,如果落库的话,关键解决的问题是数据库和缓存如何做到原子性?

加事务不行,Redis和数据库的事务不是同一个,无法保证同时成功同时失败。大家有什么更好的方案呢?

第二:如果不进行落库,那么都存储到缓存中,如何设置定时同步的策略?

怎么做到缓存数据的稳定性?

3. Confirm 确认消息

理解Confirm 消息确认机制:消息的确认,是指生产者投递消息后,如果Broker收到消息,则会给我们生产者一个应答。

生产者进行接收应答,用来确定这条消息是否正常的发送到Broker,这种方式也是消息的可靠性投递的核心保障!

Confirm确认消息流程图

蓝色:producer 生产者 红色:MQ Broker 服务器

生产者把消息发送到Broker端,Broker收到消息之后回送给producer。Confirm Listener 监听应答。

操作是异步操作,当生产者发送完消息之后,就不需要管了。Confirm Listener 监听MQ Broker的应答。

3.1 如何实现Confirm确认消息?

第一步:在channel上开启确认模式:channel.confirmSelect()

第二步;在chanel上 添加监听:addConfirmListener,监听成功和失败的返回结果,根据具体的结果对消息进行重新发送、或记录日志等后续处理!

3.2 代码编写:

生产者:

消费者:

工具类:

先启动消费端=》再启动生产端

3.3 查看管控台:

队列1

队列2

3.4 打印结果:

消费端打印结果

生产端打印结果

可以观察到消费端先接收到消息,之后生产端再接收到回调信息。如果出现磁盘已满、RabbitMQ出现异常、queue容量到达上限都可能接收到no ack

如果ack和no ack消息都未接收到,这就是之前所说的。RabbitMQ出现网络闪断,可以采用上面所说的消息补偿。

4. Return消息机制Return Listener用于处理一些不可路由的消息!

我们的消息生产者,通过指定一个Exchange和Routingkey,把消息送达到某一个队列中去,然后我们的消费者监听队列,进行消费处理操作!

但是在某些情况下,如果我们在发送消息的时候,当前的exchange不存在或者指定的路由key路由不到,这个时候如果我们需要监听这种不可达的消息,就要使用Return Listener!

在基础API中有一个关键的配置项:Mandatory:如果为true,则监听器会接收到路由不可达的消息,然后进行后续处理,如果为false,那么broker端自动删除该消息!

4.1 Return消息机制流程

Return消息机制流程

Producer生产端将消息发送到MQ Broker端,但是出现NotFind Exchange,发送的消息的Exchange,在Broker端未能找到。或者找到了,但是路由key路由不到指定的队列。因此是一个错误的消息。

这个时候,生产端应该知道发送的这条消息,并不会被处理。因此MQ Broker提供了这种Return机制,将这些不可达的消息发送给生产端,这时候生产端就需要设置Return Listener去接收这些不可达的消息。然后及时记录日志,去处理这些消息。

4.2 代码演示

生产者:

消费者:

ConnectionUtils 工具代码在上面。

启动消费端,并查看管控台。

4.3 查看管控台

Exchanges

queues

4.4 查看打印结果

放开消费端代码:channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());

消费端打印结果:

消费端打印结果

可以看到打印结果正常,此时再改代码为:

channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());

生产端打印结果

可以看到生产端接收到了不可达的消息。

文末欢迎关注个人微信公众号:Coder编程

获取最新原创技术文章和免费学习资料,更有大量精品思维导图、面试资料、PMP备考资料等你来领,方便你随时随地学习技术知识!

转载地址:https://blog.csdn.net/weixin_39965881/article/details/111813475 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!

上一篇:f2 柱状图滚动 钉钉小程序_钉钉小程序----使用阿里的F2图表
下一篇:eslint 保存自动修复_vscode配置eslint保存自动修复

发表评论

最新留言

做的很好,不错不错
[***.243.131.199]2024年02月29日 15时20分29秒

关于作者

    喝酒易醉,品茶养心,人生如梦,品茶悟道,何以解忧?唯有杜康!
-- 愿君每日到此一游!

推荐文章

一寸照纯红色底图片_Ella陈嘉桦也是“时髦精”,穿玫红色西装配拼色半身裙,高级洋气... 2019-04-21
米哈游客户端笔试题_Garena校招 游戏客户端开发通关面经 2019-04-21
airpodspro没有弹窗_使用AirPods Pro一天的主观感受 2019-04-21
创建物化视图commit_视图及范式 2019-04-21
函数传参字典_Python新手上车17:函数传递任意多个参数 2019-04-21
去掉数组最后一个元素_【一天一大 lee】在排序数组中查找元素的第一个和最后一个位置 (难度:中等) Day20201201... 2019-04-21
秦九韶算法递推公式_算法讲解之复杂度分析 2019-04-21
python房价数据分析波士顿代码数据_python数据分析-波士顿房价预测-Go语言中文社区... 2019-04-21
redis线程阻塞原因排插_Redis阻塞原因详解 2019-04-21
labview自动保存报表_基于LabVIEW的Excel报表的自动生成功能 2019-04-21
geotool 导出shp_Java 读取shape文件 2019-04-21
mysql 关联更新_MySQL UPDATE多表关联更新 2019-04-21
mysql call_mysql的call用法 call调用函数的例子 2019-04-21
python参数验证_参数验证,Python中的最佳实践 2019-04-21
python画多层网络_在pymn中修改多层网络图 2019-04-21
java net 安卓_android -------- java.net.UnknownServiceException 2019-04-21
java 密钥 aes 解密_Java中AES加密解密以及签名校验 2019-04-21
java树转化成图_Java 转换一组数据为树型数据 2019-04-21
java 底层ppt_Java 如何设置 PPT 中的形状排列方式 具体内容 2019-04-21
mysql service5.7_Mysql5.7服务下载安装 2019-04-21