3.1 消息持久化 (Message Persistence)

3.1 消息持久化 (Message Persistence)

3.1 消息持久化 (Message Persistence)

在RabbitMQ中，消息持久化是指将消息存储在磁盘上，以确保即使RabbitMQ服务器发生故障或重启，消息也不会丢失。这对于需要可靠消息传递的应用至关重要，例如金融交易、订单处理和日志记录等。

3.1.1 为什么需要消息持久化？

默认情况下，RabbitMQ将消息存储在内存中。这意味着如果服务器发生故障，所有未被消费的消息都将丢失。在某些情况下，这种行为是可以接受的，例如对于可以容忍少量数据丢失的实时监控数据。但是，对于大多数业务场景，消息丢失是不可接受的。

消息持久化提供了一种机制来确保消息在服务器故障后仍然可用。通过将消息写入磁盘，即使服务器重启，消息也可以从磁盘恢复并继续传递。

3.1.2 如何实现消息持久化

要实现消息持久化，需要在两个层面进行配置：

1. Exchange和Queue的持久化：

   • 声明Exchange和Queue时，需要将其durable属性设置为true。这告诉RabbitMQ将Exchange和Queue的元数据存储在磁盘上。

2. 消息的持久化：

   • 发布消息时，需要将消息的delivery_mode属性设置为2 (persistent)。这告诉RabbitMQ将消息内容写入磁盘。

重要提示： 仅仅将消息的delivery_mode设置为2是不够的。如果Exchange或Queue没有被声明为持久化的，那么消息仍然可能丢失。

3.1.3 代码实践

以下是使用Python (pika库) 实现消息持久化的示例代码：

import pika
def publish_message(queue_name, message):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    # 声明一个持久化的queue
    channel.queue_declare(queue=queue_name, durable=True)
    # 发送一个持久化的message
    channel.basic_publish(exchange='',
                          routing_key=queue_name,
                          body=message,
                          properties=pika.BasicProperties(
                              delivery_mode=2,  # make message persistent
                          ))
    print(" [x] Sent %r" % message)
    connection.close()
def consume_message(queue_name):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue=queue_name, durable=True)
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        # 确认消息已被处理
        ch.basic_ack(delivery_tag=method.delivery_tag)
    # 设置prefetch_count为1，避免消费者一次性获取大量消息
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(queue=queue_name, on_message_callback=callback)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()
if __name__ == '__main__':
    queue_name = 'durable_queue'
    #publish_message(queue_name, 'This is a persistent message!')
    try:
        consume_message(queue_name)
    except KeyboardInterrupt:
        print("Interrupted")
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)
​

代码解释：

• publish_message函数:

  • 连接到RabbitMQ服务器。

  • 使用channel.queue_declare(queue=queue_name, durable=True)声明一个持久化的queue，durable=True表示该queue是持久化的。

  • 使用channel.basic_publish发送消息，delivery_mode=2 表示该消息是持久化的。

• consume_message函数:

  • 连接到RabbitMQ服务器。

  • 使用channel.queue_declare(queue=queue_name, durable=True)声明一个持久化的queue，确保queue存在且为持久化。

  • channel.basic_qos(prefetch_count=1)设置prefetch_count为1，这对于确保消息的公平分发和防止消费者崩溃时消息丢失至关重要。

  • 使用channel.basic_consume开始消费消息。

  • 在回调函数中，ch.basic_ack(delivery_tag=method.delivery_tag)用于手动确认消息已被处理。 这对于持久化消息至关重要，确保消息在消费者处理失败时不会丢失。

运行步骤：

1. 安装pika库：pip install pika

2. 确保RabbitMQ服务器正在运行。

3. 运行脚本。 先注释consume_message 的调用，取消注释 publish_message 函数，运行一次， 生产消息。 然后，注释掉 publish_message 的调用，取消注释 consume_message 函数，运行，消费消息。

注意事项：

• 手动确认 (Manual Acknowledgements): 在消费者端，使用手动确认（basic_ack）是非常重要的。 如果在消费者处理消息时发生错误，并且没有发送确认，RabbitMQ会重新将消息放入队列，以便其他消费者可以处理它。这确保了消息至少被处理一次。 如果使用自动确认，消息会在发送给消费者后立即从队列中删除，即使消费者处理失败，消息也会丢失。

• prefetch_count: 设置合适的prefetch_count值也很重要。 这个值限制了消费者在收到确认之前可以接收的消息数量。 如果prefetch_count设置得太高，消费者可能会一次性获取大量消息，如果消费者崩溃，这些消息可能会丢失或需要重新传递。 将prefetch_count设置为1可以确保消费者一次只处理一条消息，从而提高可靠性。

3.1.4 持久化的工作原理

当消息被标记为持久化时，RabbitMQ会将消息写入磁盘。具体来说，消息会被写入到磁盘上的一个日志文件中。当RabbitMQ服务器重启时，它会从日志文件中恢复消息，并将它们重新放入队列中。

RabbitMQ使用一种称为"镜像队列"的机制来实现高可用性。镜像队列会将队列复制到多个RabbitMQ节点上。如果一个节点发生故障，其他节点可以接管并继续处理消息。

3.1.5 持久化的性能考虑

消息持久化会带来一定的性能开销，因为将消息写入磁盘比写入内存要慢。因此，在决定是否使用消息持久化时，需要权衡可靠性和性能之间的关系。

以下是一些可以提高持久化性能的技巧：

• 使用SSD磁盘： SSD磁盘比传统的机械硬盘具有更高的读写速度。

• 调整磁盘I/O设置： 可以调整操作系统的磁盘I/O设置来优化磁盘性能。

• 使用多个磁盘： 可以将RabbitMQ的数据目录分布在多个磁盘上，以提高I/O吞吐量。

• 批量确认： 将多个消息的确认操作批量处理可以减少I/O操作，提高性能。

3.1.6 消息持久化流程图

流程图解释：

1. Producer: 生产者发送消息到Exchange。

2. Durable Exchange?: RabbitMQ检查Exchange是否被声明为持久化。如果不是，消息可能会丢失。

3. Durable Queue?: RabbitMQ检查Queue是否被声明为持久化。如果不是，消息可能会丢失。

4. Message Properties: delivery_mode=2?: RabbitMQ检查消息的delivery_mode是否设置为2 (persistent)。如果不是，消息将仅存储在内存中。

5. Write Message to Disk: 如果Exchange、Queue和消息都被标记为持久化，RabbitMQ会将消息写入磁盘。

6. Store Message in Memory: 如果消息没有被标记为持久化，RabbitMQ会将消息存储在内存中。

7. Process Message: 消费者处理消息。

8. Ack?: 消费者发送确认（Ack）给RabbitMQ。

9. Remove Message from Queue: 如果收到确认，RabbitMQ将从队列中删除消息。

10. Message Remains in Queue for Redelivery: 如果没有收到确认，RabbitMQ会将消息保留在队列中，以便重新传递给其他消费者。

11. Message Loss: 如果Exchange或Queue没有被声明为持久化，或者消息没有被标记为持久化，那么在RabbitMQ服务器发生故障时，消息可能会丢失。

3.1.7 总结

消息持久化是RabbitMQ中一个重要的特性，可以确保消息在服务器故障后仍然可用。通过将Exchange、Queue和消息都标记为持久化，可以最大限度地减少消息丢失的风险。在实际应用中，需要根据业务需求权衡可靠性和性能之间的关系，选择合适的持久化策略。同时，结合手动确认和合理的prefetch_count设置，可以进一步提高消息传递的可靠性。