在高并发的情况下如何保证消息的可靠性?消息丢失如何解决?
本文分三部分说明
mq消息丢失场景有哪些?
创新互联-专业网站定制、快速模板网站建设、高性价比二连浩特网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式二连浩特网站制作公司更省心,省钱,快速模板网站建设找我们,业务覆盖二连浩特地区。费用合理售后完善,十年实体公司更值得信赖。
如何避免消息丢失?
大厂如何解决这些问题的?
mq消息丢失场景有哪些?
首先我们看下消息周期投递过程:
解决RabbitMQ消息丢失问题和保证消息可靠性
我们把该图分三部分,左中右三部分,每部分都会导致消息丢失情况:
1.生产者生产消息到RabbitMQ-Server 消息丢失场景
外界环境问题导致:发生网络丢包、网络故障等造成消息丢失
代码层面,配置层面,考虑不全导致消息丢失
发送端使用Confirm模式,方案不够严谨,比如MQ Server接收消息失败发送 nack给发送端后,发送端监听失败或者没做任何事情,消息丢失的情况;
再比如发送消息到exchange后,发下路由和queue没有绑定,消息会存在丢失情况,下面会讲到具体的例子。
2.RabbitMQ-Server中存储的消息丢失
消息没有持久化导致丢失
单节点或者集群模式没有镜像模式消息丢失
个别磁盘意外损害导致消息同步失败
机房被炸
3.RabbitMQ-Server到消费者消息丢失
消费者接收到相关消息之后,还没来得及处理就宕机了,消息丢失
如何避免消息丢失?
下面也是从三个方面介绍:
生产者生产消息到RabbitMQ-Server 可靠性保证
RabbitMQ-Server中存储的消息如何保证
RabbitMQ-Server到消费者消息如何不丢
1. 生产者生产消息到RabbitMQ-Server可靠性保证
这个过程,消息可能会丢,比如发生网络丢包、网络故障等造成消息丢失,一般情况下如果不采取措施,生产者无法感知消息是否已经正确无误的发送到exchange中,如果生产者能感知到的话,它可以进行进一步的处理动作,比如重新投递相关消息以确保消息的可靠性。
1.1 别担心,有一种方案可以解决:就是 AMQP协议提供的一个事务机制
RabbitMQ客户端中Channel 接口提供了几个事务机制相关的方法: channel.txSelect channel.txCommit channel.txRollback 源码截图如下:com.rabbitmq.client 包中public interface Channel extendsShutdownNotifier {}接口
在生产者发送消息之前,通过channel.txSelect开启一个事务,接着发送消息, 如果消息投递server失败,进行事务回滚channel.txRollback,然后重新发送, 如果server收到消息,就提交事务channel.txCommit但是,很少有人这么干,因为这是同步操作,一条消息发送之后会使发送端阻塞,以等待RabbitMQ-Server的回应,之后才能继续发送下一条消息,生产者生产消息的吞吐量和性能都会大大降低。
1.2 不过幸运的是RabbitMQ提供了一个改进方案,即发送方确认机制(publisher confirm)
首先生产者通过调用channel.confirmSelect方法将信道设置为confirm模式,一旦信道进入confirm模式,所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始),一旦消息被投递到所有匹配的队列之后,RabbitMQ就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一deliveryTag和multiple参数),这就使得生产者知晓消息已经正确到达了目的地了。
其实Confirm模式有三种方式实现:
串行confirm模式:producer每发送一条消息后,调用waitForConfirms()方法,等待broker端confirm,如果服务器端返回false或者超时时间内未返回,客户端进行消息重传。
批量confirm模式:producer每发送一批消息后,调用waitForConfirms()方法,等待broker端confirm。
异步confirm模式:提供一个回调方法,broker confirm了一条或者多条消息后producer端会回调这个方法。 我们分别来看看这三种confirm模式
串行confirm
批量confirm模式
上面代码是简单版本的,生产环境绝对不是循环发送的,而是根据业务情况, 各个客户端程序需要定期(每x秒)或定量(每x条)或者两者结合来pubish消息,然后等待服务器端confirm。相比普通confirm模式,批量可以极大提升confirm效率。
但是有没有发现什么问题?
问题1: 批量发送的逻辑复杂话了。
问题2: 一旦出现confirm返回false或者超时的情况时,客户端需要将这一批次的消息全部重发,这会带来明显的重复消息数量,并且,当消息经常丢失时,批量confirm性能应该是不升反降的。
异步confirm模式
异步模式需要自己多写一部分复杂的代码实现,异步监听类,监听server端的通知消息,异步的好处性能会大幅度提升,发送完毕之后,可以继续发送其他消息。 MQServer通知生产端ConfirmListener监听类:用户可以继承接口实现自己的实现类,处理消息确认机制,此处继承类代码省略,就是上面 ProxiedConfirmListener 类: 下面贴下要实现的接口:
上面的接口很有意思,如果是你的话,怎么实现? 消息投递前如何存储消息,ack 和 nack 如何处理消息?
下面看下异步confirm的消息投递流程:
解决RabbitMQ消息丢失问题和保证消息可靠性
解释下这张图片:
channerl1 连续发类1,2,3条消息到RabbitMQ-Server,RabbitMQ-Server通知返回一条通知,里面包含回传给生产者的确认消息中的deliveryTag包含了确认消息的序号,此外还有一个参数multiple=true,表示到这个序号之前的所有消息都已经得到了处理。这样客户端和服务端通知的次数就减少类,提升类性能。
加点消息存储和删除逻辑
事务机制和publisher confirm机制确保的是消息能够正确的发送至RabbitMQ,这里的“发送至RabbitMQ”的含义是指消息被正确的发往至RabbitMQ的交换器,如果此交换器没有匹配的队列的话,那么消息也将会丢失,怎么办?
这里有两个解决方案,
1. 使用mandatory 设置true
2. 利用备份交换机(alternate-exchange):实现没有路由到队列的消息
我们看下RabbitMQ客户端代码方法
Channel 类中 发布消息方法
void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException;
解释下:basicPublish 方法中的,mandatory和immediate
/** * 当mandatory标志位设置为true时,如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue, 那么会调用basic.return方法将消息返回给生产者
* 当mandatory设置为false时,出现上述情形broker会直接将消息扔掉。 */ @Setter(AccessLevel.PACKAGE) private boolean mandatory = false; /** * 当immediate标志位设置为true时,如果exchange在将消息路由到queue(s)时发现对于的queue上没有消费者, 那么这条消息不会放入队列中。 当immediate标志位设置为false时,exchange路由的队列没有消费者时,该消息会通过basic.return方法返还给生产者。 * RabbitMQ 3.0版本开始去掉了对于immediate参数的支持,对此RabbitMQ官方解释是:这个关键字违背了生产者和消费者之间解耦的特性,因为生产者不关心消息是否被消费者消费掉 */ @Setter(AccessLevel.PACKAGE) private boolean immediate;
所以为了保证消息的可靠性,需要设置发送消息代码逻辑。如果不单独形式设置mandatory=false
使用mandatory 设置true的时候有个关键点要调整,生产者如何获取到没有被正确路由到合适队列的消息呢?通过调用channel.addReturnListener来添加ReturnListener监听器实现,只要发送的消息,没有路由到具体的队列,ReturnListener就会收到监听消息。
channel.addReturnListener(new ReturnListener() { public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP .BasicProperties basicProperties, byte[] body) throws IOException { String message = new String(body); //进入该方法表示,没路由到具体的队列 //监听到消息,可以重新投递或者其它方案来提高消息的可靠性。 System.out.println("Basic.Return返回的结果是:" + message); } });
此时有人问了,不想复杂化生产者的编程逻辑,又不想消息丢失,那么怎么办? 还好RabbitMQ提供了一个叫做alternate-exchange东西,翻译下就是备份交换器,这个干什么用呢?很简单,它可以将未被路由的消息存储在另一个exchange队列中,再在需要的时候去处理这些消息。
那如何实现呢?
简单一点可以通过webui管理后台设置,当你新建一个exchange业务的时候,可以给它设置Arguments,这个参数就是 alternate-exchange,其实alternate-exchange就是一个普通的exchange,类型最好是fanout 方便管理
解决RabbitMQ消息丢失问题和保证消息可靠性
当你发送消息到你自己的exchange时候,对应key没有路由到queue,就会自动转移到alternate-exchange对应的queue,起码消息不会丢失。
下面一张图看下投递过程:
解决RabbitMQ消息丢失问题和保证消息可靠性
那么有人有个疑问,上面介绍了,两种方式处理,发送的消息无法路由到队列的方案, 如果备份交换器和mandatory参数一起使用,会有什么效果?
答案是:mandatory参数无效
总结下上面内容,主要如何保证消息从生产者到RabbitMQ Server 端可靠性
1. Transaction: 消息落盘,只能同步开启、提交及回滚。
2. Confirm:消息进入缓冲区,支持同步、异步、批量确认。
3. Transaction和publisher confirm机制两者是互斥的
4. 一般在生产者这块避免数据丢失,都是用 Confirm 机制的。
2.RabbitMQ-Server中存储的消息如何保证
一般消息都是存内存中的,如果消息没有持久化硬盘,一天机器需要重启,获取意外停电,重启机器后,消息全丢了,所以消息持久化是必备。
名称栏目:在高并发的情况下如何保证消息的可靠性?消息丢失如何解决?
网站地址:http://ybzwz.com/article/pphdcg.html