(一).为什么会有可靠性传输问题上图是RabbitMQ的消息传递图。从生产者发送消息到消费者消费消息这个过程中消息是可能会丢失的。那么我们就来看一下具体有哪几个场景会出现消息丢失的问题。①.生产者将消息发送到RabbitMQ可能会失败。如果是因为些网络问题那么就会导致RabbitMQ无法收到生产者发来的消息。②.消息在交换机中无法路由到指定的队列。当我们的代码或配置信息写错没导致交换机和队列之间无法正确的绑定此时就会导致消息路由失败那么队列就无法获取到交换机发来的消息。③.消息队列自身原因导致消息丢失。当消息到达RabbitMQ后RabbitMQ Server 挂了那么就会导致刚刚到的消息就丢失了④.消费者消费消息异常导致消息丢失。当消息到达消费者之后消费者由于自身问题导致挂了还没来得及消费此时消息就会丢失。下面就分别来介绍一下RabbitMQ对于“消息可靠性”问题做出的措施。(二).发送方确认“发送方确认”主要是用来解决当生产者发送消息之后消息到底有没有正确地到达服务器的问题的。事实上针对这个问题有两种解决方案一种是通过事务一种是通过“发送方确认”。事务后面再进行介绍这里先介绍发送方确认。关于“发送方确认”RabbitMQ提供了两个方式来控制消息的可靠性投递一种是confirm确认模式一种是return退回模式。下面进行具体介绍。1.confirm确认模式confirm确认模式指的是当生产者在发送消息的时候针对于生产者设置一个ConfirmCallBack的监听无论消息是否到达交换机这个监听都会被执行如果交换机能够成功接收则ACK设置为True如果没有收到消息ACK设置为False也就是说confirm确认模式针对的是从生产者到交换机这一阶段的消息可靠性传输。下面进行具体的实现(1).配置相关信息correlated表示的是异步回调确认当消息发送完成后异步回调ConfirmCallback消息发送不会阻塞主线程。none表示关闭生产者确认机制。simple表示的同步等待确认发送消息后阻塞等待MQ返回确认结果发送一条阻塞等回执再发送下一条。这里我们设置成correlated。(2).设置确认回调并发送消息(3).进行测试当进行测试的时候可以发现所有的结果都是符合预期的2.return退回模式return退回模式指的是当消息到达交换机之后根据路由规则把消息放入到队列中。在交换机到队列的过程中如果消息没有被任何队列消费可以选择把消息回退给生产者。当消息回退给发送方的时候我们可以设置一个返回回调方法对消息进行处理。也就是说return退回模式针对的是从交换机到消息队列这一阶段的消息可靠性传输。(1).配置相关信息(2).设置返回回调并发送消息(3).进行测试routingKey为“confirm”由于第二个消息的routingKey为“confirm111”所以无法让队列接收到消息所以只能退回可以看到队列中只有一条消息(三).持久化持久化解决的是当RabbitMQ服务停掉以后生产者发来的消息不丢失问题。RabbitMQ的持久化分为三个部分交换机的持久化队列的持久化和消息的持久化1.交换机的持久化在声明交换机的时候我们通过durable()方法然后将参数设置为true就可以将交换机设置为持久化的了。在默认情况下交换机就是持久化的即使我们不设置也是持久化的如果需要设置为非持久化那么就将durable参数设置为false。2.队列的持久化队列的持久化我们通过声明durable参数来设置的。如果队列不进行持久化则当RabbitMQ服务重启之后队列则会被删除此时数据就会丢失。事实上之前创建的队列都是持久化的通过源码可以看到声明队列的时候默认就是持久化的。如果想要设置为非持久化则可以通过nonDurable()方法进行设置3.消息的持久化消息的持久化需要把消息的投递模式设置为PERSISTENT。在进行设置的时候我们可以传一个Message对象4.综合测试(1).交换机设置为持久化和非持久化消息是否丢失当我发送消息的时候可以发现两个队列都收到了消息下面重启RabbitMQ当我重启之后交换机没有了那么消息也就不存在了。队列也没有了消息也就没有了是因为我们设置的队列是非持久化的当我将交换机设置为非持久化队列设置为持久化再进行测试没重启化RabbitMQ之前有两条数据虽然重启之后还是两条数据但是交换机不存在了所以说交换机持久化存储消息不丢失交换机非持久化存储消息也不会丢失(3).队列设置为持久化消息设置为持久化消息是否丢失在进行上面的测试的时候针对于“pers.true.queue”这个队列队列是持久化存储的消息也是持久化存储的并且通过查看测试结果可以看到消息也存储下来了所以说队列设置为持久化消息设置为持久化消息不会丢失(4).队列设置为持久化消息设置为非持久化消息是否丢失保留一个持久化的队列发送非持久化的消息可以发现已经有一条数据了下面重启RabbitMQ服务可以发现消息已经丢失了所以说队列持久化消息持久化消息不丢失队列持久化消息非持久化消息丢失(5).队列设置为非持久化消息设置为持久化和非持久化消息是否丢失重启之前可以发现是两条消息重启之后发现队列没有了那么对应的消息也就没有了所以说队列非持久化消息持久化消息丢失队列非持久化消息非持久化消息丢失(四).消息确认1.消息确认机制消息确认机制指的是当消息到达消费者之后RabbitMQ就会把这条消息删除但是消费者不一定能够正常处理消息如果消费者没有正常处理消息例如消费者挂了同时RabbitMQ已经把这条数据删除了此时就会造成数据的丢失消息确认机制就是用来解决上述问题的。也就是说消息确认机制针对的是从队列到消费者这一阶段的消息可靠性传输。消息确认机制有两种一种是自动确认。当autoAck为true的时候RabbitMQ会自动把发送出去的消息设置为确认然后从内存中删除不管消费者是否真正的消费到了消息。一种是手动确认。当autoAck为false的时候RabbitMQ会等待消费者显式地调用Basic.Ack命令回复确认信号后才从内存中移除消息 。从Web管理平台上也可以看到当前队列中Ready状态和Unacked状态的消息数2.手动确认方法RabbitMQ提供了不同的确认应答方式。消费者客户端可以调用与其对应的channel的相关方法。一共有三种①.肯定确认Channel.basicAck(long deliveryTagboolean multiple)deliveryTag表示的是消息的唯一标识他是一个单调递增的64位长整型。每个通道上的delivery是唯一的当消费者确认一条消息时必须使用对应的通道上进行确认。multiple表示是否批量确认。②.否定确认Channel.basicReject(long deliveryTag , boolean requeue)requeue表示的是当消费者拒绝后如果设置为true则RabbitMQ会将这条消息重新入队如果设置为false则RabbitMQ会将消息从队列中移除。③.否定确认Channel.basicNack(long deliveryTag , boolean multiple , boolean requeue)3.示例Spring AMQP对消息确认机制提供了三种策略NONE MANUAL AUTONONE消息一旦发送给了消费者无论消费者是否收到RabbitMQ都会自动确认消息乳沟消费者处理消息失败则消息可能会丢失。AUTO是Spring AMQP的默认方式。消费者在消息处理成功后会自动确认消息如果处理过程中出现了异常则不会确认消息MANUAL手动确认模式下必须成功处理消息后显示调用basicAck()方法来确认消息。如果消息未被确认RabbitMQ会认为消息尚未被成功处理并且会在消费者可用时重新投递该消息。(1).NONEⅠ.配置确认机制Ⅱ.生产者发送消息可以看到能够正常发送消息Ⅲ.消费者消费消息Ⅳ.测试代码当运行起来的时候发现程序报错了同时消息已经被丢弃了(2).AUTOⅠ.配置确认机制Ⅱ.生产者发送消息可以看到能够正常发送消息Ⅲ.消费者消费消息Ⅳ.测试代码当我进行测试的时候后端不断地打日志。这是因为消费者没有确认消息同时可以看到在队列中有一条Unacked的消息。当我将后端日志停掉之后发现这条消息又变成了Ready状态(3).MANUALⅠ.配置确认机制Ⅱ.生产者发送消息可以看到能够正常发送消息Ⅲ.消费者消费消息Ⅳ.代码测试由于原本在队列中有一条消息并且在处理消息的时候代码中存在异常。所以就会调用basicNack()方法然后重新发出消息所以后端一直在打日志当我们将异常注释掉之后发现已经成功处理完成了队列中也已经空了当我们将basicAck()方法注释掉之后再重新发送消息可以发现这条消息也是一直没有被处理(五).如何保证RabbitMQ消息的可靠性1.如果是生产者将消息发送到RabbitMQ失败的话我们可以采取“发送方确认的confirm模式”当MQ成功收到后会回调ack如果失败则回调nack2.如果是消息无法从交换机路由到指定队列我们可以采取“发送方确认的return回调机制”3.如果指定队列自身原因导致数据丢失我们可以采取“持久化”的方式开启队列持久化和消息持久化。如果消息过期队列满消息被拒绝消息会变成死信转发到死信队列。4.如果是消费者的原因导致消息丢失我们可以采取“消息确认”的方式默认情况下是自动应答的我们可以开启“手动确认”当业务处理成功后调用basicAck()方法通知MQ删除消息。当异常的时候可以调用basicNack()方法让消息重返队列重试。重试的时候我们可以限制重试次数如果次数达到上限则投递到死信队列。