首页 / 更多教程 / rabbit--消息持久化
rabbit--消息持久化
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了rabbit--消息持久化,小编现在分享给大家,供广大互联网技能从业者学习和参考。文章包含5131字,纯文字阅读大概需要8分钟。
内容图文
消息的可靠性是RabbitMQ的一大特色,那么RabbitMQ是如何保证消息可靠性的呢——消息持久化。
为了保证RabbitMQ在退出或者crash等异常情况下数据没有丢失,需要将queue,exchange和Message都持久化。
queue的持久化
queue的持久化是通过durable=true来实现的。
1 using (var connection = factory.CreateConnection()) 2 { 3using (var channel = connection.CreateModel()) 4 { 5 channel.QueueDeclare(queue:"hello", durable: false, exclusive: false, autoDelete: false, arguments: null); 67 } 8 }
参数说明:
queue:queue名称
exclusive:排他队列;如果一个队列被声明为排他队列,该队列仅对首次申明它的连接可见,并在连接断开时自动删除。
这里需要注意三点:1. 排他队列是基于连接可见的,同一连接的不同信道是可以同时访问同一连接创建的排他队列;
2.“首次”,如果一个连接已经声明了一个排他队列,其他连接是不允许建立同名的排他队列的,这个与普通队列不同;
3.即使该队列是持久化的,一旦连接关闭或者客户端退出,该排他队列都会被自动删除的,这种队列适用于一个客户端发送读取消息的应用场景。
autoDelete:自动删除,如果该队列没有任何订阅的消费者的话,该队列会被自动删除。这种队列适用于临时队列。
quereDeclear相关的4种方法,如下:
1 Queue.DeclareOk queueDeclare() throws IOException; 2 Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, 3 Map<String, Object> arguments) throws IOException; 4void queueDeclareNoWait(String queue, boolean durable, boolean exclusive, boolean autoDelete, 5 Map<String, Object> arguments) throws IOException; 6 Queue.DeclareOk queueDeclarePassive(String queue) throws IOException;
其中需要说明的是queueDeclarePassive(String queue)可以用来检测一个queue是否已经存在。
如果该队列存在,则会返回true;如果不存在,就会返回异常,但是不会创建新的队列。
消息的持久化
如果将queue的持久化标识为durable设置true,则代表是一个持久化的队列,那么在服务重启后,也会存在;因为服务会把持久化的queue存放在硬盘上。当服务重启的时候,会重新设置之前持久化的queue。
队列是可以被持久化,但是里面的消息是否为持久化那还要看消息的持久化设置。
也就是说重启之前那个queue里面还没有发出去的消息的话,重启之后那队列里面是不是还存在原来的消息,这个就要取决于发生着在发送消息时对消息的设置了。
如果要在重启后保持消息的持久化必须设置消息是持久化的标识。
设置消息的持久化:
channel.BasicPublish(exchange: "test", routingKey: "task_queue", basicProperties: MessageProperties.PERSISTENT_TEXT_PLAIN, body: body);
参数解析:
exchange:exchange名字
routingKey:routingKey名字
body:表示发送的消息体
basicProperties:首先看下BasicProperties的定义
1 public BasicProperties( 2 String contentType,//消息类型如:text/plain 3 String contentEncoding,//编码 4 Map<String,Object> headers, 5 Integer deliveryMode,//1:nonpersistent 2:persistent 6 Integer priority,//优先级 7 String correlationId, 8 String replyTo,//反馈队列 9 String expiration,//expiration到期时间10 String messageId, 11 Date timestamp, 12 String type, 13 String userId, 14 String appId, 15 String clusterId)
deliveryMode=1表示不持久化;deliveryMode=2则表示持久化。
那么MessageProperties.PERSISTENT_TEXT_PLAIN又是什么鬼?
1 public static final BasicProperties PERSISTENT_TEXT_PLAIN = 2new BasicProperties("text/plain", 3null, 4null, 52, 60, null, null, null, 7null, null, null, null, 8null, null);
可以看到这其实就是讲deliveryMode设置为2的BasicProperties的对象,为了方便编程而出现的一个东东
另一种实现:
1 AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); 2 builder.deliveryMode(2); 3 AMQP.BasicProperties properties = builder.build(); 4 channel.basicPublish("exchange.persistent", "persistent",properties, "persistent_test_message".getBytes());
设置了队列和消息的持久化之后,当broker服务重启的之后,消息依旧存在。单只设置队列持久化,重启之后消息会丢失;单只设置消息的持久化,重启之后队列消失,既而消息也丢失。单单设置消息持久化而不设置队列的持久化显得毫无意义。
exchange的持久化
上面阐述了队列的持久化和消息的持久化,如果不设置exchange的持久化对消息的可靠性来说没有什么影响,但是同样如果exchange不设置持久化,那么当broker服务重启之后,exchange将不复存在,那么既而发送方rabbitmq producer就无法正常发送消息。同样设置exchange的持久化。exchange的持久化设置也特别简单,方法如下:
channel.ExchangeDeclare(exchange: "test", type: "direct/topic/header/fanout", durable: true);即在声明的时候讲durable字段设置为true即可。
1 Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable) throws IOException; 2 Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, 3 Map<String, Object> arguments) throws IOException; 4Exchange.DeclareOk exchangeDeclare(String exchange, String type) throws IOException; 5Exchange.DeclareOk exchangeDeclare(String exchange, 6 String type, 7 boolean durable, 8 boolean autoDelete, 9 boolean internal, 10 Map<String, Object> arguments) throws IOException; 11void exchangeDeclareNoWait(String exchange, 12 String type, 13 boolean durable, 14 boolean autoDelete, 15 boolean internal, 16 Map<String, Object> arguments) throws IOException; 17 Exchange.DeclareOk exchangeDeclarePassive(String name) throws IOException;
引申的问题
1.将queue,exchange, message等都设置了持久化之后就能保证100%保证数据不丢失了嚒?
答案是否定的。
首先,从consumer端来说,如果这时autoAck=true,那么当consumer接收到相关消息之后,还没来得及处理就crash掉了,那么这样也算数据丢失,这种情况也好处理,只需将autoAck设置为true(方法定义如下),然后在正确处理完消息之后进行手动ack(channel.basicAck).
其次,关键的问题是消息在正确存入RabbitMQ之后,还需要有一段时间(这个时间很短,但不可忽视)才能存入磁盘之中,RabbitMQ并不是为每条消息都做fsync的处理,可能仅仅保存到cache中而不是物理磁盘上,在这段时间内RabbitMQ broker发生crash, 消息保存到cache但是还没来得及落盘,那么这些消息将会丢失。
那么这个怎么解决呢?首先可以引入RabbitMQ的mirrored-queue即镜像队列,这个相当于配置了副本,当master在此特殊时间内crash掉,可以自动切换到slave,这样有效的保障了HA, 除非整个集群都挂掉,这样也不能完全的100%保障RabbitMQ不丢消息,但比没有mirrored-queue的要好很多,很多现实生产环境下都是配置了mirrored-queue的。还有要在producer引入事务机制或者Confirm机制来确保消息已经正确的发送至broker端。RabbitMQ的可靠性涉及producer端的确认机制、broker端的镜像队列的配置以及consumer端的确认机制,要想确保消息的可靠性越高,那么性能也会随之而降,鱼和熊掌不可兼得,关键在于选择和取舍。
2.消息什么时候刷到磁盘?
写入文件前会有一个Buffer,大小为1M,数据在写入文件时,首先会写入到这个Buffer,如果Buffer已满,则会将Buffer写入到文件(未必刷到磁盘)。
有个固定的刷盘时间:25ms,也就是不管Buffer满不满,每隔25ms,Buffer里的数据及未刷新到磁盘的文件内容必定会刷到磁盘。
每次消息写入后,如果没有后续写入请求,则会直接将已写入的消息刷到磁盘:使用Erlang的receive x after 0实现,只要进程的信箱里没有消息,则产生一个timeout消息,而timeout会触发刷盘操作。
原文:http://www.cnblogs.com/cuijl/p/8073346.html
内容总结
以上是互联网集市为您收集整理的rabbit--消息持久化全部内容,希望文章能够帮你解决rabbit--消息持久化所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错,欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 gblab@vip.qq.com 举报,一经查实,本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。