四种途径提升RabbitMQ传输数据的可靠性
前言
RabbitMQ虽然有对队列及消息等的一些持久化设置,但其实光光只是这一个是不能够保障数据的可靠性的,下面我们提出这样的质疑:
(1)RabbitMQ生产者是不知道自己发布的消息是否已经正确达到服务器呢,如果中间发生网络异常等情况呢?消息必然会丢失!
(2)RabbitMQ如果没有设置队列持久化,RabbitMQ服务器重后队列的元数据会丢失,消息自然也会丢失!
(3)RabbitMQ如果消费者设置自动确认,即autoAck为true,那么不管消费者发生什么情况,该消息会自动从队列中移除,实际上消费者有可能挂掉,消息必然会丢失!
(4)RabbitMQ中的消息如果没有匹配到队列时,那么消息也会丢失!
本文其实也就是结合以上四个方面进行讲解的,主要参考《RabbitMQ实战指南》(有需要PDF电子书的可以评论或者私信我),本文截图也来自其中,另外可以对一些RabbitMQ的概念的认识可以参考我的上两篇博文认识RabbitMQ交换机模型、RabbitMQ是如何运转的?
一、设置mandotory参数、AE备份交换器
针对前言中的第(4)个问题,我们可以通过设置mandotory参数与AE备份交换器来解决
1、mandotory参数
1)当为true时,交换器无法根据自身的类型和路由键找到一个符合条件的队列,此时RabbitMQ会调用Basic.Return命令将消息返回给生产者,消息将不会丢失
2)当为false时,消息将会被直接丢弃。
3)RabbitMQ通过addReturnListener添加ReturnLisener监听器监听获取没有被正确路由到合适队列的消息。
channel.basicPublish(EXCHANGE NAME, "", true, MessageProperties.PERSISTENT_TEXT_PLAIN, "mandatory test".getBytes()); 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);
}
}); .
2、AE备份交换器
Alternate Exchange,简称AE,不设置mandatory参数,那么消息将会被丢失,设置mandatory参数的话,需要添加ReturnListner监听器,增加复杂代码,如果既不想增加代码又不想消息丢失,则使用AE,将没有被路由的消息存储于RabbitMQ中。当mandatory参数用AE一起使用时,mandatory将失效。在介绍AE之前,也认识RabbitMQ对于消息的过期时间TTL设置以及队列的过期时间TTL设置
2.1 TTL过期时间设置
可以对队列设置TTL与消息设置TTL,其中消息设置TTL经常用于死信队列、延迟队列等高级应用中。
1)设置消息TTL
设置TTL过期时间一般有两种当时:一是通过队列属性,对队列中所有消息设置相同的TTL。二就是对消息本身单独设置,每条消息TTL不同。如果一起使用时候,TTL小的为准,当一旦超过设置的TTL时间时,就会变成“死信”。
方式一:针对每条消息设置TTL是通过增加expiration的属性参数实现的,不可能像方式二一样扫描整个队列再判断是否过期,只有当该消息即将被消费时再判定是否过期即可删除,也就是消息即使已经过期,但不一定立马被删除!
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); // 持久化消息 builder deliveryMode(2); // 设置 TTL=60000ms builder expiration( 60000 ); AMQP.BasicProperties properties = builder. build(); channel.basicPublish(exchangeName, routingKey, mandatory, properties, "ttlTestMessage".getBytes());
方式二:通过队列属性设置消息TTL是增加x-message-ttl参数实现的,只需要扫描整个队列头部即可立即删除,也就是消息一旦过期就会被删除!
Map<String, Object> argss = new HashMap<String , Object>(); argss.put("x-message-ttl", 6000); channel.queueDeclare(queueName, durable, exclusive, autoDelete, argss) ;
2)设置队列TTL
通过在队列中添加参数x-message-ttl参数实现,设置队列被自动删除前处于未被使用状态的时间,注意是队列的使用状态,并不是消息是否被消费的状态
设置ttl=30min的队列,时间一到RabbitMQ会保证队列被删除,但是不会保证删除的速度有多快。
Map<String, Object> args = new HashMap<String, Object>{); args.put("x-expires", 1800000); channel.queueDeclare("myqueue", false, false, false, args);
2.2 AE备份交换器的使用
声明交换器的时候,添加alternate-exchange参数实现,或通过策略实现。前者优先级高。从代码角度需要以下三个步骤,具体代码如下:
Map<String, Object> args = new HashMap<String, Object>(); args.put("a1ternate-exchange", "myAe"); channe1.exchangeDec1are("norma1Exchange", "direct", true, fa1se, args); channe1.exchangeDec1are("myAe", "fanout", true, fa1se, nu11) ; channe1.queueDec1are( "norma1Queue", true, fa1se, fa1se, nu11); channe1.queueB nd("norma1Queue", "norma1Exchange", "norma1Key"); channe1.queueDec1are("unroutedQueue", true, fa1se, fa1se, nu11);
1)声明normalExchange类型为direct的交换器、类型为fanout的myAe备份交换器;并且normalExchange的备份交换器为myAe(备份交换器建议使用fanout类型交换器)
2)声明normalQueue队列,声明unrouteQueue队列;
3)通过路由键normalKey绑定normalExchange与normalQueue,不适用路由键绑定unrouteQueue与myAe