如何保障消息100%投递成功？如何保证消息幂等性？

可能出现消息丢失的情况&＃xff1f;

1. Producer在把Message发送Broker的过程中&＃xff0c;因为网络问题等发生丢失&＃xff0c;或者Message到了Broker&＃xff0c;但是出了问题&＃xff0c;没有保存下来

“

针对这个问题&＃xff0c;Producer可以开启MQ的事务&＃xff0c;如果这个过程出现异常&＃xff0c;进行回滚&＃xff0c;但是有个很大的问题&＃xff0c;你提交一个事务就会阻塞在那&＃xff0c;非常影响性能&＃xff0c;生产环境肯定不会开启事务&＃xff0c;一般都是使用confirm机制

1. Broker接收到Message暂存到内存&＃xff0c;Consumer还没来得及消费&＃xff0c;Broker挂掉了

“

可以通过持久化设置去解决&＃xff1a;(1) 创建Queue的时候设置持久化&＃xff0c;保证Broker持久化Queue的元数据&＃xff0c;但是不会持久化Queue里面的消息; (2) 将Message的deliveryMode设置为2&＃xff0c;可以将消息持久化到磁盘&＃xff0c;这样只有Message支持化到磁盘之后才会发送通知Producer ack。

这两步过后&＃xff0c;即使Broker挂了&＃xff0c;Producer肯定收不到ack的&＃xff0c;就可以进行重发。

1. Consumer有消费到Message&＃xff0c;但是内部出现问题&＃xff0c;Message还没处理&＃xff0c;Broker以为Consumer处理完了&＃xff0c;只会把后续的消息发送

“

这时候&＃xff0c;就要关闭autoack&＃xff0c;消息处理过后&＃xff0c;进行手动ack

生产端如何保证可靠性投递&＃xff1f;

1. 保证消息的成功发出

2. 保证MQ节点的成功接收

3. 发送端收到MQ节点&＃xff08;Broker&＃xff09;的确认应答

4. 完善的消息补偿机制

解决方案

消息落库

1. 消息落库&＃xff0c;对消息状态进行变更&＃xff0c;对于高并发环境下数据库压力很大&＃xff0c;因为需要写多次数据库

   

整体流程&＃xff1a;

1. 业务数据和消息都进行落库

2. 生产端发送message给Broker

3. Broker给Confirm响应返回生产端

4. 接收到confirm&＃xff0c;对message状态更改

5. 分布式定时任务获取消息的状态

6. 如果消息不能成功投递&＃xff0c;重新进行发送&＃xff0c;记录重发次数

7. 当重发3次之后&＃xff0c;将状态修改&＃xff0c;只能人工进行干预

消息的延迟投递

消息的延迟投递&＃xff0c;做二次确认&＃xff0c;回调检查。适合高并发环境&＃xff0c;减少写库的次数

整体流程&＃xff1a;

1. 上游服务首先将业务代码入库&＃xff0c;发送message给Broker

2. 发送第二个延迟确认消息

3. 下游服务监听消息进行消费

4. 发送确认消息&＃xff0c;这里不是confirm机制&＃xff0c;而是一条新的消息

5. 通过回调服务监听这个confirm消息&＃xff0c;然后把消息进行入库

6. 回调服务检查到延迟确认消息&＃xff0c;会在数据库查询是否有这条消息

7. 如果没有查到这条消息&＃xff0c;回调服务通过RPC给一个重新发送命令到上游系统

“

相比第一种方案&＃xff0c;这里减少了一次message入库&＃xff0c;confirm机制是消息可靠性投递的一个核心。

RabbitMQ可能导致出现非幂等性的情况

1. 可靠性消息投递机制&＃xff1a;consumer回复confirm出现网络闪断&＃xff0c;producer没有收到ack&＃xff0c;定时任务轮询可能就会重新发送消息&＃xff0c;这样consumer就会收到两条消息

2. MQ Broker与消费端传输消息的过程出现网络抖动

3. 消费端故障或异常

kafka可能出现非幂等性的情况

在Consumer端offset没有提交的时候&＃xff0c;Consumer重启了&＃xff0c;这时候就会出现重复消费的情况

解决方案

唯一ID&＃43;指纹码

整体实现相对简单&＃xff0c;需要进行数据库写入&＃xff0c;利用数据库主键去重&＃xff0c;使用ID进行分库分表算法路由&＃xff0c;从单库的幂等性到多库的幂等性

1. 这里唯一ID一般就是业务表的主键&＃xff0c;比如商品ID

2. 指纹码&＃xff1a;每次操作都要生成指纹码&＃xff0c;可以用时间戳&＃43;业务编号&＃43;...组成&＃xff0c;目的是保证每次操作都是正常的

整体流程&＃xff1a;

1. 需要一个统一ID生成服务&＃xff0c;为了保证可靠性&＃xff0c;上游服务也要有个本地ID生成服务&＃xff0c;然后发送消息给Broker

2. 需要ID规则路由组件去监听消息&＃xff0c;先入库&＃xff0c;如果入库成功&＃xff0c;证明没有重复&＃xff0c;然后发给下游&＃xff0c;如果发现库里面有了这条消息&＃xff0c;就不发给下游