rabbitmq保证消息可靠 消息不丢失
三端可靠
- 发送方和mq保证消息送达到mq
- mq保证保存的消息不丢失
- 消费方和mq一起保证消息被成功消费
发送方和mq保证消息送达到mq
方案一、rabbitmq如果是用spring boot提供的模版接口发送 需要调用rabbitTemplate.convertSendAndReceive()方法发送 这个是当消息成功到队列了才会返回结果 如果失败则会抛异常 不过这就会导致等待时间比较长 适合高可靠场景
不过一般在业务开发都是完成业务以后再发消息 比如插入订单表一笔订单 发送订单创建的消息 这两步是需要保证原子性的 要么都成功要么都失败 rabbitmq支持事务消息 不过如果出现下面情况
- 开启事务
- 插入订单表
- 发送mq消息
- 提交数据库事务成功
- 提交mq事务失败
- 消息丢失
所以如果是用rabbitmq的事务消息来做 其实在极端情况是会丢失消息的 在这里可以采用一个异步命令组件提供的方案https://github.com/bojiw/asyncmd
- 开启事务
- 插入订单表
- 插入异步命令表
- 提交数据库事务
- 线程扫描异步命令表捞取消息
- 通过rabbitTemplate.convertSendAndReceive()方法发送
- 如果失败 则重试 并且报警
方案二、如采用rabbitTemplate.convertAndSend和confirms(消费回调)加Return(错误回调)模式
- convertAndSend 发送到mq 立刻返回 不管交换机是否成功处理 所以并发会高
- confirms(消费回调) 实现接口ConfirmCallback 消息成功发送到rabbitmq交换机上则会回调接口 入参ack为true代表成功发送到交换机 false代表异常
- Return(错误回调) 实现接口ReturnCallback 消息从交换机到队列 成功不会回调 如果发送到队列失败 则会调用回调
上面这种方式如果在回调中处理消息发送失败的逻辑时出现异常或者应用服务器挂了 则会导致消息丢失 因为只会回调一次
这种情况可以采用加一张消息表 先插入消息表 然后扫表发送消息 confirms回调成功 则更新表状态 如果回调的时候异常 则消息表会重新发送 这种就会出现消息重发的情况 不过一般消息消费者都要保证幂等 所以这个问题不大 不过如果出现以下情况
- 数据库有两个字段 confirms默认0 和 return 默认0
- 回调成功confirms=1 回调失败confirms=2 错误回调return=2
- 当回调成功 confirms=1 错误回调处理失败没有成功更新表 则return还是0
- 这个时候你扫表就不确定需不需要重发消息 因为如果消息成功到队列 表的状态也是confirms=1 return=0
- 无法对发送队列成功和发送队列失败可在回调异常这两种情况做区分
这里逻辑就会出问题 所以只能处理消息成功到交换机 是否到队列则不管 因为一般都是成功的 除了极端情况 比如队列被人误删除
方案二和方案一其实从整个流程来讲 发送消息速度其实差不多的 不过可靠性还是方案一高一点
mq保证保存的消息不丢失
消息、交换机、队列都需要设置持久化
消费方和mq一起保证消息被成功消费
消费者开启手动确认
acknowledge="manual"
在业务代码里 成功处理业务 才返回给rabbitmq消费成功的确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
如果业务处理失败则重新放到队列重新消费
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
由消费者 只有业务成功处理才进行ack 记得需要做好幂等
不过这里rabbitmq在重试这块没有做好 如果不确定会一直重试 如果因为依赖的一个系统挂了 要一个小时以后才会启动成功 在这一个小时里会一直重试 这就会对rabbitmq和消费者带来一定的压力 这块也可以采用异步命令组件提供的方案https://github.com/bojiw/asyncmd
- 接收消息
- 把消息插入异步命令
- 返回rabbitmq成功
- 异步组件执行业务逻辑
- 调用接口失败重试
- 重试一定次数则不重试 由人工进行处理 也可以把重试间隔设置的长一点 比如前三次每隔1s重试 第四次隔一个小时重试
