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

可能出现消息丢失的情况?

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

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

  • Broker接收到Message暂存到内存,Consumer还没来得及消费,Broker挂掉了

可以通过持久化设置去解决:

(1) 创建Queue的时候设置持久化,保证Broker持久化Queue的元数据,但是不会持久化Queue里面的消息;

(2) 将Message的deliveryMode设置为2,可以将消息持久化到磁盘,这样只有Message支持化到磁盘之后才会发送通知Producer ack。

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

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

这时候,就要关闭autoack,消息处理过后,进行手动ack

生产端如何保证可靠性投递?

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

解决方案

消息落库

  • 消息落库,对消息状态进行变更,对于高并发环境下数据库压力很大,因为需要写多次数据库
  如何保障消息100%投递成功?如何保证消息幂等性 image.png

整体流程

  1. 业务数据和消息都进行落库
  2. 生产端发送message给Broker
  3. Broker给Confirm响应返回生产端
  4. 接收到confirm,对message状态更改
  5. 分布式定时任务获取消息的状态
  6. 如果消息不能成功投递,重新进行发送,记录重发次数
  7. 当重发3次之后,将状态修改,只能人工进行干预

消息的延迟投递

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

  如何保障消息100%投递成功?如何保证消息幂等性 image.png

 

整体流程

  1. 上游服务首先将业务代码入库,发送message给Broker
  2. 发送第二个延迟确认消息
  3. 下游服务监听消息进行消费
  4. 发送确认消息,这里不是confirm机制,而是一条新的消息
  5. 通过回调服务监听这个confirm消息,然后把消息进行入库
  6. 回调服务检查到延迟确认消息,会在数据库查询是否有这条消息
  7. 如果没有查到这条消息,回调服务通过RPC给一个重新发送命令到上游系统

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

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

  1. 可靠性消息投递机制:consumer回复confirm出现网络闪断,producer没有收到ack,定时任务轮询可能就会重新发送消息,这样consumer就会收到两条消息
  2. MQ Broker与消费端传输消息的过程出现网络抖动
  3. 消费端故障或异常

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

  如何保障消息100%投递成功?如何保证消息幂等性 image.png

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

解决方案

唯一ID+指纹码

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

  1. 这里唯一ID一般就是业务表的主键,比如商品ID
  2. 指纹码:每次操作都要生成指纹码,可以用时间戳+业务编号+...组成,目的是保证每次操作都是正常的
  如何保障消息100%投递成功?如何保证消息幂等性 image.png

整体流程

  1. 需要一个统一ID生成服务,为了保证可靠性,上游服务也要有个本地ID生成服务,然后发送消息给Broker
  2. 需要ID规则路由组件去监听消息,先入库,如果入库成功,证明没有重复,然后发给下游,如果发现库里面有了这条消息,就不发给下游

 

上一篇:Kafka工作原理


下一篇:Java开发五年,kafka查看topic数据内容