2.4 消息 (Message)

2.4 消息 (Message)

2.4 RabbitMQ 核心概念详解：消息 (Message)

在RabbitMQ消息队列系统中，消息（Message）是信息传递的基本单元。它承载着从生产者到消费者的实际数据，是构建异步通信和解耦系统的核心要素。理解消息的结构、属性以及如何在RabbitMQ中操作消息至关重要。

2.4.1 消息的定义与作用

消息，简单来说，就是生产者发送给RabbitMQ服务器，最终被消费者接收和处理的数据载体。在消息队列的语境下，消息不仅仅是数据本身，还包含了关于如何处理这些数据的元数据信息。

消息在RabbitMQ中扮演着以下关键角色：

• 数据载体: 消息的核心作用是携带实际的应用数据，例如订单信息、日志数据、用户请求等。

• 异步通信媒介: 消息使得生产者和消费者之间可以异步通信。生产者无需等待消费者的响应，只需将消息发送到RabbitMQ，即可继续执行其他任务。消费者则在需要时从队列中获取消息进行处理。

• 系统解耦工具: 消息队列通过消息的传递，将生产者和消费者解耦。生产者无需知道消费者的具体位置和状态，只需关注消息的发送。消费者也无需关心消息的来源，只需关注消息的处理。这种解耦性提高了系统的灵活性和可维护性。

• 流量削峰与缓冲: 在高并发场景下，消息队列可以作为缓冲层，平滑流量高峰。生产者可以将大量的消息快速发送到队列中，消费者则可以按照自身处理能力，逐步从队列中获取消息进行处理，避免系统瞬间过载。

2.4.2 消息的结构：属性 (Properties) 和 负载 (Payload)

一个RabbitMQ消息主要由两部分组成：属性 (Properties) 和 负载 (Payload)。

1. 属性 (Properties)

消息属性是附加在消息上的元数据信息，用于描述消息的特征、路由规则以及其他控制信息。属性以键值对的形式存在，并被包含在消息的头部。RabbitMQ预定义了一系列标准属性，同时也允许用户自定义属性。

常见的RabbitMQ消息属性包括：

• delivery_mode (投递模式): 指定消息的持久性。

  • 1 (Transient): 瞬时消息，RabbitMQ服务器重启后消息会丢失。

  • 2 (Persistent): 持久消息，RabbitMQ服务器会将消息持久化到磁盘，即使服务器重启，消息也不会丢失。但持久化会降低消息的吞吐量。

  • 重要性: 决定消息的可靠性。对于重要数据，应设置为持久消息。

• content_type (内容类型): 描述消息负载的MIME类型，例如 text/plain, application/json 等。

  • 作用: 消费者可以根据 content_type 来正确解析消息负载。

• content_encoding (内容编码): 描述消息负载的编码方式，例如 gzip, utf-8 等。

  • 作用: 消费者可以根据 content_encoding 来正确解码消息负载。

• priority (优先级): 消息的优先级，取值范围通常为 0-9，数字越大优先级越高。

  • 作用: 优先级高的消息会优先被投递给消费者。但RabbitMQ默认的队列实现并不严格支持优先级队列，需要启用优先级队列插件。

• correlation_id (关联ID): 用于关联请求和响应消息。在RPC（远程过程调用）场景中，生产者可以使用 correlation_id 标记请求消息，消费者在响应消息中携带相同的 correlation_id，生产者可以根据此ID将响应消息与原始请求关联起来。

  • 应用场景: 请求-响应模式。

• reply_to (回复队列): 指定消费者处理完消息后，应将响应消息发送到哪个队列。通常与 correlation_id 配合使用。

  • 应用场景: 请求-响应模式。

• expiration (过期时间): 消息的过期时间，单位为毫秒。如果消息在队列中停留时间超过过期时间，RabbitMQ会自动删除该消息。

  • 作用: 防止队列中积压过多的过期消息，节省资源。

• message_id (消息ID): 消息的唯一标识符，由生产者生成。

  • 作用: 用于消息追踪和去重。

• timestamp (时间戳): 消息的发送时间，由生产者设置。

  • 作用: 用于消息的时序分析和监控。

• type (消息类型): 消息的类型标签，用于区分不同类型的消息。

  • 作用: 消费者可以根据 type 字段来区分和处理不同类型的消息。

• user_id (用户ID): 发送消息的用户ID，用于安全审计和权限控制。

• app_id (应用ID): 发送消息的应用ID，用于区分消息来源。

• cluster_id (集群ID): 集群环境下的集群ID，用于标识消息所属的集群。

• headers (自定义头部): 一个键值对字典，允许用户自定义消息属性。

  • 作用: 扩展消息属性，满足特定的业务需求。例如，可以添加版本号、业务类型、路由键等自定义信息。

2. 负载 (Payload)

消息负载是消息体，包含了实际要传输的数据内容。负载可以是任何二进制数据，RabbitMQ不对负载的内容做任何限制和解析。常见的负载类型包括：

• 文本数据 (Text): 例如 JSON, XML, CSV 等格式的文本数据。

• 二进制数据 (Binary): 例如图片、音频、视频、序列化后的对象等。

生产者需要根据实际需求选择合适的负载类型，并在消息属性中设置 content_type 和 content_encoding，以便消费者能够正确解析和处理负载数据。

2.4.3 消息的生命周期

消息从生产者发送到消费者接收，经历以下生命周期阶段：

1. 生产者发布消息 (Publish Message): 生产者将消息发送到指定的交换机 (Exchange)。在发布消息时，生产者需要指定交换机名称、路由键 (Routing Key) 和消息属性 (Properties) 以及消息负载 (Payload)。

2. 交换机路由消息 (Route Message): 交换机接收到消息后，根据交换机类型 (Direct, Fanout, Topic, Headers) 和路由键，将消息路由到一个或多个绑定的队列 (Queue)。

3. 队列存储消息 (Store Message): 队列接收到路由过来的消息，并将消息存储在队列中，等待消费者消费。

4. 消费者消费消息 (Consume Message): 消费者订阅队列，从队列中获取消息进行处理。消费者可以选择主动拉取消息 (Pull/Get) 或被动接收消息 (Push/Consume)。

5. 消费者确认消息 (Acknowledge Message): 消费者成功处理消息后，需要向RabbitMQ服务器发送确认 (Acknowledgement) 信号。RabbitMQ收到确认信号后，才会将消息从队列中删除。如果消费者在处理消息过程中发生异常，或者未发送确认信号，RabbitMQ会将消息重新投递给其他消费者或重新放回队列，以便重新消费 (取决于是否启用消息重试机制和消息的配置)。

2.4.4 消息的代码实践 (Python Pika 示例)

以下使用 Python 的 Pika 客户端库，演示如何创建和发送带有属性的消息，以及如何接收和处理消息并获取消息属性。

1. 生产者代码 (producer.py):

import pika
import json
import time
# 连接 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明交换机 (Direct Exchange)
exchange_name = 'direct_exchange_example'
channel.exchange_declare(exchange=exchange_name, exchange_type='direct')
# 路由键
routing_key = 'order.created'
# 消息负载 (JSON 格式)
message_payload = {
    'order_id': 'ORDER-001',
    'customer_id': 'CUST-123',
    'item': 'Product A',
    'quantity': 2
}
message_body = json.dumps(message_payload)
# 消息属性
properties = pika.BasicProperties(
    delivery_mode=2,  # 消息持久化
    content_type='application/json',
    content_encoding='utf-8',
    priority=5,
    correlation_id=str(time.time()), # 使用时间戳作为关联ID
    reply_to='response_queue', # 设置回复队列
    expiration='30000', # 消息过期时间 30 秒
    message_id='MSG-001',
    timestamp=int(time.time()),
    type='order_event',
    headers={'version': '1.0', 'source': 'order_service'} # 自定义头部
)
# 发布消息到交换机
channel.basic_publish(exchange=exchange_name,
                      routing_key=routing_key,
                      body=message_body,
                      properties=properties)
print(f" [x] Sent message with routing key: {routing_key}")
connection.close()
​

代码详解:

• 连接 RabbitMQ: 使用 pika.BlockingConnection 创建到 RabbitMQ 服务器的连接。

• 声明交换机: 使用 channel.exchange_declare 声明一个 Direct 类型的交换机 direct_exchange_example。

• 消息负载: 创建一个 Python 字典 message_payload，并使用 json.dumps 将其转换为 JSON 字符串作为消息负载。

• 消息属性: 创建 pika.BasicProperties 对象，设置各种消息属性，例如 delivery_mode (持久化), content_type, priority, correlation_id, reply_to, expiration, message_id, timestamp, type 以及自定义的 headers。

• 发布消息: 使用 channel.basic_publish 将消息发布到指定的交换机，并指定路由键、消息负载和消息属性。

2. 消费者代码 (consumer.py):

import pika
import json
# 连接 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明交换机 (与生产者相同)
exchange_name = 'direct_exchange_example'
channel.exchange_declare(exchange=exchange_name, exchange_type='direct')
# 声明队列
queue_name = 'order_queue'
channel.queue_declare(queue=queue_name)
# 绑定队列到交换机，并指定路由键
channel.queue_bind(exchange=exchange_name,
                   queue=queue_name,
                   routing_key='order.created')
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
    print(" [x] Received message:")
    print(f"     Delivery Tag: {method.delivery_tag}")
    print(f"     Exchange: {method.exchange}")
    print(f"     Routing Key: {method.routing_key}")
    print("     Properties:")
    print(f"         Delivery Mode: {properties.delivery_mode}")
    print(f"         Content Type: {properties.content_type}")
    print(f"         Content Encoding: {properties.content_encoding}")
    print(f"         Priority: {properties.priority}")
    print(f"         Correlation ID: {properties.correlation_id}")
    print(f"         Reply-to: {properties.reply_to}")
    print(f"         Expiration: {properties.expiration}")
    print(f"         Message ID: {properties.message_id}")
    print(f"         Timestamp: {properties.timestamp}")
    print(f"         Type: {properties.type}")
    print(f"         Headers: {properties.headers}")
    print("     Payload:")
    payload = json.loads(body.decode('utf-8')) # 解析 JSON 负载
    print(json.dumps(payload, indent=4))
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag) # 手动确认消息
# 设置消息消费回调函数，并开启消费者
channel.basic_consume(queue=queue_name,
                      on_message_callback=callback,
                      auto_ack=False) # 关闭自动确认，使用手动确认
channel.start_consuming()
​

代码详解:

• 连接 RabbitMQ 和声明交换机/队列: 与生产者代码类似，连接 RabbitMQ 服务器，并声明相同的交换机和队列。

• 绑定队列: 使用 channel.queue_bind 将队列 order_queue 绑定到 direct_exchange_example 交换机，并指定路由键 order.created。

• 消费回调函数 callback: 定义消息消费的回调函数。当消费者收到消息时，该函数会被调用。

  • 打印消息信息: 在 callback 函数中，打印接收到的消息的各种信息，包括 delivery_tag, exchange, routing_key 以及消息属性 properties 中的各个字段。

  • 解析负载: 使用 json.loads 解析 JSON 格式的消息负载。

  • 手动确认消息: 使用 ch.basic_ack(delivery_tag=method.delivery_tag) 手动确认消息。

• 设置消费者并启动消费: 使用 channel.basic_consume 设置消费者，指定队列名、回调函数和关闭自动确认 ( auto_ack=False )。然后使用 channel.start_consuming() 启动消费者，开始监听队列并接收消息。

运行示例:

1. 启动 RabbitMQ 服务器。

2. 运行 producer.py 脚本，发送消息。

3. 运行 consumer.py 脚本，接收并处理消息。

在消费者终端，你将看到接收到的消息的详细信息，包括消息属性和负载内容。

2.4.5 消息属性的最佳实践

• 合理设置 delivery_mode: 根据业务需求选择消息的持久性。对于重要数据，务必设置为持久消息。但也要注意持久化会影响性能。

• 使用 content_type 和 content_encoding: 明确指定消息负载的类型和编码，方便消费者正确解析。

• 利用 correlation_id 和 reply_to: 在需要请求-响应模式的场景中，使用这两个属性实现消息的关联和回复。

• 设置 expiration 防止消息堆积: 对于有时效性的消息，设置合理的过期时间，避免队列中堆积过多的无效消息。

• 自定义 headers 扩展消息属性: 根据业务需求，灵活使用自定义头部，添加额外的元数据信息。

• 谨慎使用 priority: 优先级队列会增加系统的复杂度，并且并非所有队列实现都完美支持优先级。应谨慎评估是否真的需要优先级特性。

• 消息大小限制: RabbitMQ 默认的消息大小限制约为 128MB。应避免发送过大的消息，必要时可以考虑消息分片或使用消息存储服务。

2.4.6 总结

消息 (Message) 是 RabbitMQ 的核心概念之一。深入理解消息的结构、属性和生命周期，对于构建可靠、高效的消息队列系统至关重要。合理利用消息属性，可以实现消息的持久化、路由、优先级控制、关联性以及其他高级特性，从而满足各种复杂的业务场景需求。通过代码实践和不断学习，可以更好地掌握 RabbitMQ 消息的使用技巧，构建更健壮的消息驱动应用。