Java架构直通车——MQ消息的可靠性投递方案
什么是生产端的可靠性投递
生产端要做到可靠性投递,需要以下几点:
- 保障消息的成功发出
- 保障MQ节点的成功接收
- 发送端收到MQ节点(Broker)的确认应答
- 完善消息的补偿机制:也就是上面三个步骤失败了的补偿机制。
生产端可靠性投递的解决方案:
- 消息落库,对消息状态进行打标
在消息发送的时候,把消息持久化到数据库中,然后消息有个状态:比如说,刚发送的时候可能说消息叫做发送中,收到回送响应后把消息状态进行变更,比如说打标为已发送。
那么对于没有响应的消息,可以进行轮询操作,抓取没有成功的消息进行重新发送,做一个最大努力尝试的次数即可。
- step1:进行数据的入库,包括对业务的入库和对消息的入库,比如说订单服务,包括对订单业务数据的存储(保存订单消息或者创建订单)以及对订单消息的存储。如果这两个持久化操作抛出异常,快速失败即可。
- step2:producer发送消息给broker。
- step3:broker应答结果给producer。Confirm Listener异步的去监听Broker回送的这条响应。
- step4:更新数据库的消息状态,表示确认消息发送成功。
- step5:使用分布式定时任务去抓取数据库中不成功的消息(一般来讲使用分布式的任务,防止重复抓取)。
- step6:消息重发,设置重发次数。
- step7:消息重发次数过多,表明消息投递失败。这时候可以给出一个补偿系统去获取这些消息失败的原因(当然也可以人工去操作)。
这种方式有个缺点:
在数据持久化的时候, 需要经过两次磁盘写操作(业务消息入库和消息信息入库)。并且后续需要update。(一般来说不需要事务,事务会造成严重的性能瓶颈)。在高并发场景下,这种解决方案不合适。
- 消息的延迟投递,做消息的二次确认、回调检查
这种方式是比较好的可靠性投递机制。这种方式主要是为了减少数据库持久化的操作。
- step1:upstream service也就是上游服务(生产端),首先会将业务数据持久化(一般来说不加事务),落库完了之后把消息发到broker。
- step2:第二条消息是延迟消息投递检查,设置一个消息发送的延迟时间,比如设置5min,延迟5min后发送消息。这里的延迟消息虽然和第一次放松的消息在业务上说的是一个事情,但是投递的队列不是一个队列。
- step3:downstream service也就是下游服务(消费端),收到消息并处理。
- step4:下游服务消息处理完成之后,向broker发送一个confirm的消息,这个confirm消息也是一个消息队列。也就是处理成功的消息会被编辑成为一个新的消息投递到broker。
- step5:callback service也就是回调服务充当了之前的Message DB的角色,回调服务去监听下游服务发送的消息,然后回调服务做消息的持久化服务。
- step 6:回调服务监听延迟消息的队列,5min后延迟消息发送到该队列,回调服务监听到了该消息,并在Message DB数据库中进行检查,发现DB里收到了这个数据,不需要组任何事情。如果发现没有查到这个message,那么进行一个RPC消息,通知上游再次发送这个消息。
这里是异步做了消息的持久化操作,这个回调服务就相当于一个补偿服务,在主流程中减少了一次数据持久化操作,由回调服务实现。