Rabbitmq可靠性消息投递两种方案(定时任务版本+延迟队列版本)

- 日理万妓 2023-02-17 02:15 17阅读 0赞

# Rabbitmq可靠性消息投递 两种方案 #

每天多学一点点~  
话不多说，这就开始吧…

### 文章目录 ###

*  Rabbitmq可靠性消息投递 两种方案
 *  1.前言
 *  2.定时任务版本+人工补偿
 *  3.延迟队列版本+人工补偿
 *  4.结语

# 1.前言 #

平时工作中用到的最后的消息队列就是rabbitmq。听说阿里云上还有amqp？哦，不错哟~这里提供两种可靠性消息的提供方案。在此记录一下！

# 2.定时任务版本+人工补偿 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70]

若真实的生产环境上， Biz\_db(业务) 和 msg\_db(消息) 是两个数据库，需要加 jta 监听，会有性能问题

**消息状态**

*  0 发送中
 *  1 mq服务端签收消息成功
 *  2 mq服务端签收消息失败
 *  3 消费端消费消息成功
 *  4 消费端消费消息失败

**-----上游服务 开始 ------**

1.  生产者(order服务) 消息如入库 状态为0 Biz\_db(业务) 和 msg\_db(消息) 生产上一般要做两个数据库，存在分布式事务问题(jta解决)  
    因为是两个不同的数据库，所以存在 分布式事务(跨库事务) JTA 但性能会低
2.  Product(order服务) 发送消息去 Mq-server
3.  Mq-server 回调 消息可路由
4.  判断 消息 是否可达  
    4.1 消息可达  
    ConfirmListener消息确定模式，判断ack  
    4.1.1 ack为true 更新msg\_db(消息)状态为 1(mq服务签收消息成功)  
    4.1.2 ack为false 更新msg\_db(消息)状态为 2(mq服务签收失败)
    
    4.2 消息不可达  
    ReutrnListener处理消息不可达的消息 去 msg\_db(消息) 更新消息状态为 2(mq服务签收失败)

**-----上游服务 结束 ------**

**-----下游服务 开始 ------**

1.  Mq-server发送消息给 er(product商品服务) 进行消费
2.  Consumer(product商品服务) 判断消息 是否 异常 回调  
    6.1 Y异常 发送nack 更新 msg\_db(消息)状态为 4(消息端签收失败) 是否要重回队列，看业务需求
    
    6.2 N正常 发送ack 更新 msg\_db(消息)状态为 3(消息端签收消息)
3.  定时抓取:消息表5分钟之内，消息状态不是3(消息端签收消息 也就是成功的)的消息

**-----下游服务 结束 ------**

# 3.延迟队列版本+人工补偿 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 1]  
**消息状态**

*  0 发送中
 *  1 mq服务端签收消息成功
 *  2 mq服务端签收消息失败
 *  3 消费端消费消息成功
 *  4 消费端消费消息失败

1.  Product(order订单服务)业务数据biz\_db入库
2.  Product(order订单服务)发送消息去MQ-SERVER **这一步 失败 可以用 confirmListener 但是这一步失败的原因比较少**
3.  Product(order订单服务)发送延迟检查消息 可以与第二步间隔5min，具体根据业务需求
4.  Consumer(product库存服务)监听业务消息 即监听 第二步 发送的消息
5.  Consumer(product库存服务) 消费成功之后，发送确认消息 至 MQ-SERVER ，一定要比比 第三步 要快
6.  Callback(回调检查服务) 监听第五步 确认消费消息
7.  Callback(回调检查服务) 入库 第五步 发送的确认消费消息 msg\_db
8.  监听 第三步 发送的 延迟 检查消息 去msg\_db查(通过消息id进行关联的) 若没查到，说明第五步 失败，去第九步延迟对比，重新发送
9.  延迟检查对比 代码中没对比 直接跳到10 重新发送了
10. 延迟检查失败，回调Product(order订单服务)服务重新发送消息

**比起第一种方案**：

*  **优势** 第一步不需要入msg\_db，不需要加分布式事务jta，性能显著提高
 *  **劣势** 增加了3,5两个队列 和 一个工程（Callback回调检查回复）

# 4.结语 #

**世上无难事，只怕有心人，每天积累一点点，fighting！！！**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200610184820139.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200610185348942.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw==,size_16,color_FFFFFF,t_70