Rabbitmq消息持久化

消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保证消息可靠性的呢——消息持久化。?
为了保证RabbitMQ在退出或者crash等异常情况下数据没有丢失，需要将queue，exchange和Message都持久化。

queue的持久化

queue的持久化是通过durable=true来实现的。?
一般程序中这么使用：

/**

* amqp_queue_declare

*

* @param [in] state connection state – TCP连接

* @param [in] channel the channel to do the RPC on –信道

* @param [in] queue queue –队列名称

* @param [in] passive passive

* @param [in] durable durable –是否持久化

* @param [in] exclusive exclusive –是否排他

* @param [in] auto_delete auto_delete –是否自动删除

* @param [in] arguments arguments

* @returns amqp_queue_declare_ok_t

*/

AMQP_PUBLIC_FUNCTION

amqp_queue_declare_ok_t *

AMQP_CALL
amqp_queue_declare(amqp_connection_state_t
state, amqp_channel_t
channel, amqp_bytes_t
queue, amqp_boolean_t
passive, amqp_boolean_t
durable, amqp_boolean_t
exclusive, amqp_boolean_t
auto_delete, amqp_table_t
arguments)

{

amqp_queue_declare_t
req;

req.ticket = 0;

req.queue = queue;

req.passive = passive;

req.durable = durable;

req.exclusive = exclusive;

req.auto_delete = auto_delete;

req.nowait = 0;

req.arguments = arguments;

?

return
amqp_simple_rpc_decoded(state, channel, AMQP_QUEUE_DECLARE_METHOD, AMQP_QUEUE_DECLARE_OK_METHOD, &req);

}

@param [in] durable 在申明一个队列时，有一个参数amqp_boolean_t
durable用来指定该队列中的消息是否持久化；

@param [in] exclusive 排他队列，如果一个队列被声明为排他队列，该队列仅对首次申明它的连接可见，并在连接断开时自动删除。这里需要注意三点：1. 排他队列是基于连接可见的，同一连接的不同信道是可以同时访问同一连接创建的排他队列；2."首次"，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同；3.即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的，这种队列适用于一个客户端发送读取消息的应用场景；

@param [in] auto_delete 自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列；

消息的持久化

如果将queue的持久化标识durable设置为true,则代表是一个持久的队列，那么在服务重启之后，也会存在，因为服务会把持久化的queue存放在硬盘上，当服务重启的时候，会重新创建之前被持久化的queue。队列是可以被持久化，但是里面的消息是否为持久化那还要看消息的持久化设置。也就是说，重启之前那个queue里面还有没有发出去的消息的话，重启之后那队列里面是不是还存在原来的消息，这个就要取决于发生着在发送消息时对消息的设置了。?
如果要在重启后保持消息的持久化必须设置消息的持久化的标识；

amqp_basic_properties_t
props;

props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;

props.content_type = amqp_cstring_bytes("text/plain");

props.delivery_mode = 2; /* persistent delivery mode */

die_on_error(amqp_basic_publish(conn,

1,

amqp_cstring_bytes(exchange),

amqp_cstring_bytes(routingkey),

0,

0,

&props,

amqp_cstring_bytes("test message")),

"Publishing");

发送消息之前要先设置一个消息属性结构体amqp_basic_properties_t， delivery_mode指定消息的持久化设置；

delivery_mode=2将消息设置为持久化模式，队列中这条消息将会被持久化保存；

设置了队列和消息的持久化之后，当broker服务重启的之后，消息依旧存在。单只设置队列持久化，重启之后消息会丢失；单只设置消息的持久化，

重启之后队列消失，既而消息也丢失。单单设置消息持久化而不设置队列的持久化显得毫无意义。

exchange的持久化

上面阐述了队列的持久化和消息的持久化，如果不设置exchange的持久化对消息的可靠性来说没有什么影响，但是同样如果exchange不设置持久化，

那么当broker服务重启之后，exchange将不复存在，那么既而发送方rabbitmq producer就无法正常发送消息。这里博主建议，同样设置exchange的持久化。

exchange的持久化设置也特别简单；

/**

* amqp_exchange_declare

*

* @param [in] state connection state

* @param [in] channel the channel to do the RPC on

* @param [in] exchange exchange

* @param [in] type type

* @param [in] passive passive

* @param [in] durable durable

* @param [in] auto_delete auto_delete

* @param [in] internal internal

* @param [in] arguments arguments

* @returns amqp_exchange_declare_ok_t

*/

AMQP_PUBLIC_FUNCTION

amqp_exchange_declare_ok_t *

AMQP_CALL
amqp_exchange_declare(amqp_connection_state_t
state, amqp_channel_t
channel, amqp_bytes_t
exchange, amqp_bytes_t
type, amqp_boolean_t
passive, amqp_boolean_t
durable, amqp_boolean_t
auto_delete, amqp_boolean_t
internal, amqp_table_t
arguments)

{

amqp_exchange_declare_t
req;

req.ticket = 0;

req.exchange = exchange;

req.type = type;

req.passive = passive;

req.durable = durable;

req.auto_delete = auto_delete;

req.internal = internal;

req.nowait = 0;

req.arguments = arguments;

?

return
amqp_simple_rpc_decoded(state, channel, AMQP_EXCHANGE_DECLARE_METHOD, AMQP_EXCHANGE_DECLARE_OK_METHOD, &req);

}

参数和队列创建的类似不做详细解释；

进一步讨论

1.将queue，exchange, message等都设置了持久化之后就能保证100%保证数据不丢失了嚒？?
答案是否定的。?
首先，从consumer端来说，如果这时autoAck=true，那么当consumer接收到相关消息之后，还没来得及处理就crash掉了，那么这样也算数据丢失，这种情况也好处理，只需将autoAck设置为false(方法定义如下)，然后在正确处理完消息之后进行手动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的事务机制或者Confirm机制可以参考：RabbitMQ之消息确认机制（事务+Confirm）；幸亏本文的主题是讨论RabbitMQ的持久化而不是可靠性，不然就一发不可收拾了。RabbitMQ的可靠性涉及producer端的确认机制、broker端的镜像队列的配置以及consumer端的确认机制，要想确保消息的可靠性越高，那么性能也会随之而降，鱼和熊掌不可兼得，关键在于选择和取舍。

2.消息什么时候刷到磁盘？?
写入文件前会有一个Buffer,大小为1M,数据在写入文件时，首先会写入到这个Buffer，如果Buffer已满，则会将Buffer写入到文件（未必刷到磁盘）。?
有个固定的刷盘时间：25ms,也就是不管Buffer满不满，每个25ms，Buffer里的数据及未刷新到磁盘的文件内容必定会刷到磁盘。?
每次消息写入后，如果没有后续写入请求，则会直接将已写入的消息刷到磁盘：使用Erlang的receive x after 0实现，只要进程的信箱里没有消息，则产生一个timeout消息，而timeout会触发刷盘操作。

?

?

Rabbitmq消息持久化

原文地址：https://www.cnblogs.com/skiing886/p/8446174.html

时间： 2024-10-09 02:59:27

Rabbitmq消息持久化的相关文章

rabbitmq 消息持久化

rabbitmq 消息持久化 2016-02-18 11:19 224人阅读 评论(0) 收藏 举报  分类: 综合(15)  版权声明:本文为博主原创文章,未经博主允许不得转载. 二: 任务分发 &消息持久化 启用多个接收端的时候如果某一个receive 关闭要保证消息有反馈是否收到 send端 #-*- coding: UTF-8 -*-import pikacred = pika.PlainCredentials('zxl','pwd') #账号密码params = pika.Connec

RabbitMQ原理与相关操作(三)消息持久化

现在聊一下RabbitMQ消息持久化: 问题及方案描述 1.当有多个消费者同时收取消息,且每个消费者在接收消息的同时,还要处理其它的事情,且会消耗很长的时间.在此过程中可能会出现一些意外,比如消息接收到一半的时候,一个消费者死掉了. 这种情况要使用消息接收确认机制,可以执行上次宕机的消费者没有完成的事情. 2.在默认情况下,我们程序创建的消息队列以及存放在队列里面的消息,都是非持久化的.当RabbitMQ死掉了或者重启了,上次创建的队列.消息都不会保存. 这种情况可以使用RabbitMQ提供的消

RabbitMQ(三):消息持久化策略

原文:RabbitMQ(三):消息持久化策略 一.前言 在正常的服务器运行过程中,时常会面临服务器宕机重启的情况,那么我们的消息此时会如何呢?很不幸的事情就是,我们的消息可能会消失,这肯定不是我们希望见到的结果.所以我们希望AMQP服务器崩溃了也可以将消息恢复,这称之为消息持久化.RabbitMQ自然存在这种策略可以帮助我们完成这件事情. 二.持久化的消息 当RabbitMQ服务器重启后,原先的队列和交换器会随同里面的消息一同消失.原因在于每个队列和交换器都有durable属性,该属性默认是fa

rabbitmq 消息安全接收与消息持久化

可插拔式:一个插件,安装和写在不影响主程序运行 durable=True 持久,持续地 | 队列持久化 delivery_mode=2 消息持久化 import pika import time credentials = pika.PlainCredentials('alex', 'alex123') connection = pika.BlockingConnection(pika.ConnectionParameters( '192.168.14.52',credentials=crede

(转)RabbitMQ消息队列（三）：任务分发机制

在上篇文章中,我们解决了从发送端(Producer)向接收端(Consumer)发送“Hello World”的问题.在实际的应用场景中,这是远远不够的.从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法. 当有Consumer需要大量的运算时,RabbitMQ Server需要一定的分发机制来balance每个Consumer的load.试想一下,对于web application来说,在一个很多的HTTP request里是没有时间来处理复杂的运算的,只能通过后台的一些工作线程

RabbitMQ消息队列（三）：任务分发机制

在上篇文章中,我们解决了从发送端(Producer)向接收端(Consumer)发送“Hello World”的问题.在实际的应用场景中,这是远远不够的.从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法. 当有Consumer需要大量的运算时,RabbitMQ Server需要一定的分发机制来balance每个Consumer的load.试想一下,对于web application来说,在一个很多的HTTP request里是没有时间来处理复杂的运算的,只能通过后台的一些工作线程

(转)RabbitMQ消息队列（九）：Publisher的消息确认机制

在前面的文章中提到了queue和consumer之间的消息确认机制:通过设置ack.那么Publisher能不到知道他post的Message有没有到达queue,甚至更近一步,是否被某个Consumer处理呢?毕竟对于一些非常重要的数据,可能Publisher需要确认某个消息已经被正确处理. 在我们的系统中,我们没有是实现这种确认,也就是说,不管Message是否被Consume了,Publisher不会去care.他只是将自己的状态publish给上层,由上层的逻辑去处理.如果Message

(转)RabbitMQ消息队列（七）：适用于云计算集群的远程调用(RPC)

在云计算环境中,很多时候需要用它其他机器的计算资源,我们有可能会在接收到Message进行处理时,会把一部分计算任务分配到其他节点来完成.那么,RabbitMQ如何使用RPC呢?在本篇文章中,我们将会通过其它节点求来斐波纳契完成示例. 1. 客户端接口 Client interface 为了展示一个RPC服务是如何使用的,我们将创建一段很简单的客户端class. 它将会向外提供名字为call的函数,这个call会发送RPC请求并且阻塞知道收到RPC运算的结果.代码如下: [python] vie

RabbitMQ消息队列随笔

本文权当各位看官对RabbitMQ的基本概念以及使用场景有了一定的了解,如果你还对它所知甚少或者只是停留在仅仅是听说过,建议你先看看这篇文章,在对RabbitMQ有了基本认识后,我们正式开启我们的RabbitMQ之旅吧,希望本文能够帮助大家在实际用到消息队列时有所帮助,如有表述的不当之处,还望各位看官指正. 一.消息队列的安装 1. RabbitMQ是用Erlang编程语言进行开发,所以首先得在Erlang官网下载Erlang运行的环境,如图,选择所需对应文件进行下载,并进行安装: 2. 设置环