RabbitMQ消息丢失如何解决

本篇文章为大家展示了RabbitMQ消息丢失如何解决 ，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。首先我们看下消息周期投递过程：我们把该图分三部分，左中右，每部分都会导致消息丢失情况,下面就详细聊聊每个阶段消息是如何丢的：1） 外界环境问题导致：发生网络丢包、网络故障等造成RabbitMQ Server端收不到消息，因为生产环境的网络是很复杂的，网络抖动，丢包现象很常见，下面会讲到针对这个问题是如何解决的。2） 代码层面，配置层面，考虑不全导致消息丢失事例1:
一般情况下，生产者使用Confirm模式投递消息，如果方案不够严谨，比如RabbitMQ Server 接收消息失败后会发送nack消息通知生产者，生产者监听消息失败或者没做任何事情，消息存在丢失风险；事例2:
生产者发送消息到exchange后，发送的路由和queue没有绑定，消息会存在丢失情况，下面会讲到具体的例子，保证意外情况的发生，即使发生，也在可控范围内。1）消息未完全持久化，当机器重启后，消息会全部丢失，甚至Queue也不见了假如：你仅仅持久化了Message,而Exchange,Queue没有 香港云主机持久化，这个持久化是无效的。 记得之前公司有一哥们忘记持久化Queue导致机器重启后，Queue不见了，自然Message也丢失了。2）单节点模式问题，如果某个节点挂了，消息就不能用了,业务可能瘫痪，只能等待如果做了消息持久化方案，消息会持久化硬盘，机器重启后消息不会丢失；但是还有一个极端情况，这台服务器磁盘突然坏了（公司遇到过磁盘问题还是很多的），消息持久化不了，非高可用状态，这个模式生产环境慎重考虑。3）普通集群模式：某个节点挂了，该节点上的消息不能用，有影响的业务瘫痪，只能等待节点恢复重启可用（建立在消息持久化）虽然这个模式进步了一点点，多个节点，但是消息还是不能保证可靠，为什么呢？因为RabbitMQ 集群模式有点特殊，队列的内容仅仅存在某一个节点上面，不会存在所有节点上面，所有节点仅仅存放消息结构和元数据（可以理解为索引，这也是为了提高性能，如果每次把所有内容同步到所有节点是有开销代价的）。 下面自己画了一张图介绍普通集群丢失消息情况： 这里有三个节点，通常情况下一个磁盘节点，两个内存节点，首先先说明下， Queue1 内容仅仅存在节点note1上面，在创建队列的时候已经固定了，note2,note3 仅仅存放的是元数据，这个一定要清楚，Producer发送消息到note2，note2 会同步元数据到其他节点，内容会同步note1。那我们想下，图中的Q1问题，note1挂了，这个节点的Queues全部暂时不可用，节点恢复后可用。我们说下图片中备注2中的问题，Producer发送消息到note2，note2在同步note1前note1挂了，此时你的心情是怎么样的。。。后面会讲具体的策略4）镜像模式：可以解决上面的问题，但是还是有意外情况发生比如：持久化的消息，保存到硬盘过程中，当前队列节点挂了，存储节点硬盘又坏了，消息丢了，怎么办？下面会详细介绍消费端接收到相关消息之后，消费端还没来得及处理消息，消费端机器就宕机了，此时消息如果处理不当会有丢失风险，后面会讲到如何处理这个情况，消费端也有ack机制下面也是从三个方面介绍：1.生产者生产消息到RabbitMQ Server 可靠性保证2.RabbitMQ Server中存储的消息如何保证3.RabbitMQ Server到消费者消息如何不丢这个过程，消息可能会丢，比如发生网络丢包、网络故障等造成消息丢失，一般情况下如果不采取措施，生产者无法感知消息是否已经正确无误的发送到exchange中,如果生产者能感知到的话，它可以进行进一步的处理动作，比如重新投递相关消息以确保消息的可靠性。1.1 通常有一种方案可以解决：就是 AMQP协议提供的一个事务机制RabbitMQ客户端中Channel 接口提供了几个事务机制相关的方法：channel.txSelectchannel.txCommitchannel.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条)或者两者结合来publish消息，然后等待服务器端confirm。相比普通confirm模式，批量可以极大提升confirm效率。但是有没有发现什么问题？问题1: 批量发送的逻辑复杂化了。问题2: 一旦出现confirm返回false或者超时的情况时，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量，并且当消息经常丢失时，批量confirm性能应该是不升反降的。异步confirm模式异步模式需要自己多写一部分复杂的代码实现，异步监听类，监听server端的通知消息，异步的好处性能会大幅度提升，发送完毕之后，可以继续发送其他消息。 MQServer通知生产端ConfirmListener监听类：用户可以继承接口实现自己的实现类，处理消息确认机制，此处继承类代码省略，就是上面 ProxiedConfirmListener 类： 下面贴下要实现的接口：上面的接口很有意思，如果是你的话，怎么实现？ 消息投递前如何存储消息，ack 和 nack 如何处理消息？下面看下异步confirm的消息投递流程：解释下这张图片：channel1 连续发类1，2，3条消息到RabbitMQ-Server，RabbitMQ-Server通知返回一条通知，里面包含回传给生产者的确认消息中的deliveryTag包含了确认消息的序号，此外还有一个参数multiple=true，表示到这个序号之前的所有消息都已经得到了处理。这样客户端和服务端通知的次数就减少类，提升类性能。channel3 发送的消息失败了，生产端需要对投递消息重新投递，需要额外处理代码。 那么生产端需要做什么事情呢？因为是异步的，生产端需要存储消息然后根据server通知的消息，确认如何处理，于是我们面临的问题是：第一：发送消息之前把消息存起来第二：监听ack 和 nack 并做响应处理那么怎么存储呢？我们分析下，可以使用SortedMap 存储，保证有序，但是有个问题高并发情况下， 每秒可能几千甚至上万的消息投递出去，消息的ack要等几百毫秒的话，放内存可能有内存溢出的风险。所以建议采用KV存储，KV存储承载高并发能力高，性能好，但是要保证KV 高可用，单个有个缺点就是又引入了第三方中间件，复杂度升高。解决了上面的问题，下面还会遇到一个问题，消息丢失的另一个情况？事务机制和publisher confirm机制确保的是消息能够正确的发送至RabbitMQ，这里的“发送至RabbitMQ”的含义是指消息被正确的发往至RabbitMQ的交换器，如果此交换器没有匹配的队列的话，那么消息也将会丢失，怎么办？这里有两个解决方案，我们看下RabbitMQ客户端代码方法Channel 类中 发布消息方法解释下：basicPublish 方法中的，mandatory和immediate所以为了保证消息的可靠性，需要设置发送消息代码逻辑。如果不单独形式设置mandatory=false使用mandatory 设置true的时候有个关键点要调整，生产者如何获取到没有被正确路由到合适队列的消息呢？通过调用channel.addReturnListener来添加ReturnListener监听器实现，只要发送的消息，没有路由到具体的队列，ReturnListener就会收到监听消息。此时有人问了，不想复杂化生产者的编程逻辑，又不想消息丢失，那么怎么办？ 还好RabbitMQ提供了一个叫做alternate-exchange东西，翻译下就是备份交换器，这个干什么用呢？很简单，它可以将未被路由的消息存储在另一个exchange队列中，再在需要的时候去处理这些消息。那如何实现呢？简单一点可以通过webui管理后台设置，当你新建一个exchange业务的时候，可以给它设置Arguments，这个参数就是 alternate-exchange，其实alternate-exchange就是一个普通的exchange，类型最好是fanout 方便管理当你发送消息到你自己的exchange时候，对应key没有路由到queue，就会自动转移到alternate-exchange对应的queue，起码消息不会丢失。下面一张图看下投递过程：上述内容就是RabbitMQ消息丢失如何解决 ，你们学到知识或技能了吗？如果还想学到更多技能或者丰富自己的知识储备，欢迎关注开发云行业资讯频道。

相关推荐: CSS中block级和inline级对象有什么区别

本文小编为大家详细介绍“CSS中block级和inline级对象有什么区别”，内容详细，步骤清晰，细节处理妥当，希望这篇“CSS中block级和inline级对象有什么区别”文章能帮助大家解决疑惑，下面跟着小编的思路慢慢深入，一起来学习新知识吧。CSS中blo…