3.9 消息属性 (Message Properties)

文档摘要

3.9 消息属性 (Message Properties) RabbitMQ 消息属性 (Message Properties) 详解与实践 3.9 消息属性 (Message Properties) 的重要性消息属性是伴随消息一起发送的元数据，它们提供了关于消息本身的额外信息，以及如何处理消息的指示。这些属性允许生产者向消费者传递上下文信息，控制消息的路由、持久化、优先级、过期时间等行为，甚至可以实现更高级的消息处理模式。消息属性的重要性体现在以下几个方面：消息路由与过滤: 属性可以被 Exchange 和 Queue 用于更精细的消息路由策略。例如，根据消息的或自定义的属性，将消息路由到不同的队列，实现内容类型路由或基于属性的路由。

3.9 消息属性 (Message Properties)

RabbitMQ 消息属性 (Message Properties) 详解与实践

3.9 消息属性 (Message Properties) 的重要性

消息属性是伴随消息一起发送的元数据，它们提供了关于消息本身的额外信息，以及如何处理消息的指示。这些属性允许生产者向消费者传递上下文信息，控制消息的路由、持久化、优先级、过期时间等行为，甚至可以实现更高级的消息处理模式。

消息属性的重要性体现在以下几个方面：

消息路由与过滤: 属性可以被 Exchange 和 Queue 用于更精细的消息路由策略。例如，根据消息的 content_type 或自定义的 headers 属性，将消息路由到不同的队列，实现内容类型路由或基于属性的路由。
消息质量保证 (QoS): delivery_mode 属性控制消息的持久化，确保消息在 RabbitMQ 服务器重启后依然能够被传递，提升消息的可靠性。
消息处理控制: expiration 属性设定消息的过期时间，防止消息在队列中无限期堆积。 priority 属性允许消费者优先处理高优先级的消息。 correlation_id 和 reply_to 属性支持请求-响应模式的消息交互。
消息上下文传递: user_id, app_id, headers 等属性允许生产者传递应用层面的上下文信息，方便消费者进行业务逻辑处理和审计跟踪。
协议兼容性: content_type 和 content_encoding 属性定义了消息体的格式和编码，确保不同系统之间能够正确解析和处理消息。

简而言之，消息属性赋予了消息更丰富的含义和更灵活的处理方式，是构建复杂消息驱动应用的基础。

3.9.1 核心消息属性详解

RabbitMQ 消息属性封装在 BasicProperties 类 (Java 客户端) 或类似的结构中 (不同客户端语言有不同的实现，但概念一致)。以下列举并详细解释一些最常用的核心消息属性：

1. content_type (内容类型)

描述: MIME 类型，指示消息体的格式。例如 text/plain, application/json, image/jpeg 等。
用途: 消费者可以根据 content_type 属性来决定如何解析和处理消息体。例如，如果 content_type 是 application/json，消费者会将其解析为 JSON 对象。
默认值: text/plain
代码示例 (Python - pika):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='property_exchange', exchange_type='topic')
channel.queue_declare(queue='property_queue')
channel.queue_bind(exchange='property_exchange', queue='property_queue', routing_key='property.test')
properties = pika.BasicProperties(content_type='application/json')
channel.basic_publish(exchange='property_exchange',
                      routing_key='property.test',
                      body='{"message": "Hello, JSON!"}',
                      properties=properties)
print(" [x] Sent message with content_type: application/json")
connection.close()

2. content_encoding (内容编码)

描述: 指示消息体使用的字符编码。例如 UTF-8, gzip 等。
用途: 消费者可以根据 content_encoding 属性来解码消息体。例如，如果 content_encoding 是 gzip，消费者会先解压缩消息体。
默认值: 无默认值，通常默认为 UTF-8 或与系统默认编码一致。
代码示例 (Java):


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.AMQP;
public class PropertyProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare("property_exchange", "topic");
            channel.queueDeclare("property_queue", false, false, false, null);
            channel.queueBind("property_queue", "property_exchange", "property.test");
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .contentEncoding("gzip")
                    .build();
            String message = "This is a message that will be gzipped.";
            // 实际应用中，需要对 message 进行 gzip 压缩后再发送
            channel.basicPublish("property_exchange", "property.test", properties, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent message with content_encoding: gzip");
        }
    }
}

3. headers (消息头)

描述: 一个键值对的字典 (或 Map)，允许用户自定义属性。
用途: headers 提供了极大的灵活性，可以用于：
- 自定义路由规则: Exchange 可以基于 headers 进行路由。 Header Exchange 就是专门基于 headers 进行路由的 Exchange 类型。
- 应用层元数据: 传递业务相关的元数据，例如消息来源、消息版本、操作类型等。
- 消息过滤: 消费者可以根据 headers 属性过滤消息。
默认值: 空字典 (或空 Map)
代码示例 (Python - pika):


properties = pika.BasicProperties(headers={'x-message-source': 'order-service',
                                           'x-message-version': '1.0'})
channel.basic_publish(exchange='property_exchange',
                      routing_key='property.test',
                      body='Hello with headers!',
                      properties=properties)

4. delivery_mode (投递模式)

描述: 指示消息是否持久化。
- 1 (Transient): 消息仅保存在内存中，RabbitMQ 服务重启或崩溃时会丢失。
- 2 (Persistent): 消息会被写入磁盘，即使 RabbitMQ 服务重启或崩溃也能恢复，保证消息的可靠性。
用途: 确保消息的持久性，防止消息丢失。对于重要的业务消息，建议设置为 2 (Persistent)。
默认值: 1 (Transient)
代码示例 (Java):


AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .deliveryMode(2) // 设置为持久化
        .build();
channel.basicPublish("property_exchange", "property.test", properties, message.getBytes("UTF-8"));

5. priority (优先级)

描述: 消息的优先级，整数值，范围通常为 0-9，数字越大优先级越高。
用途: 允许消费者优先处理高优先级的消息。需要注意的是，优先级队列需要启用优先级队列插件才能生效。默认情况下，RabbitMQ 不支持优先级队列。
默认值: 无默认值，通常为 0
代码示例 (Python - pika):


properties = pika.BasicProperties(priority=5) # 设置优先级为 5
channel.basic_publish(exchange='property_exchange',
                      routing_key='property.test',
                      body='Priority message',
                      properties=properties)

6. correlation_id (关联 ID)

描述: 用于关联请求和响应的 ID。通常在请求-响应模式中使用。
用途: 在请求-响应模式中，生产者发送请求消息时设置 correlation_id，消费者处理请求后，在响应消息中也设置相同的 correlation_id，生产者可以通过 correlation_id 将响应消息与原始请求消息关联起来。
默认值: 无默认值，通常由生产者生成唯一 ID。
代码示例 (Java - 请求-响应模式):

生产者 (请求):


String correlationId = java.util.UUID.randomUUID().toString();
String replyQueueName = channel.queueDeclare().getQueue(); // 声明一个临时队列用于接收响应
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .correlationId(correlationId)
        .replyTo(replyQueueName)
        .build();
channel.basicPublish("request_exchange", "request.key", properties, requestMessage.getBytes("UTF-8"));

消费者 (响应):


DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
            .correlationId(delivery.getProperties().getCorrelationId())
            .build();
    String response = "处理结果";
    channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
};
channel.basicConsume(requestQueueName, false, deliverCallback, consumerTag -> { });

7. reply_to (回复队列)

描述: 指定消费者处理完消息后，将响应消息发送到哪个队列。通常与 correlation_id 配合使用，实现请求-响应模式。
用途: 在请求-响应模式中，生产者在请求消息中设置 reply_to 属性，指定消费者将响应消息发送到哪个队列。
默认值: 无默认值，通常由生产者指定一个临时队列或预定义的响应队列。
代码示例 (见 correlation_id 代码示例)

8. expiration (过期时间)

描述: 消息的过期时间，单位是毫秒。
用途: 设置消息的生存时间，如果消息在队列中超过了 expiration 时间仍未被消费，RabbitMQ 会自动删除该消息。可以防止消息在队列中长期堆积，占用资源。
默认值: 无默认值，消息永不过期，除非队列本身设置了消息 TTL (Time-To-Live)。
代码示例 (Python - pika):


properties = pika.BasicProperties(expiration='60000') # 设置过期时间为 60 秒 (60000 毫秒)
channel.basic_publish(exchange='property_exchange',
                      routing_key='property.test',
                      body='This message will expire in 60 seconds',
                      properties=properties)

9. message_id (消息 ID)

描述: 消息的唯一标识符。
用途: 用于消息追踪、去重等场景。可以由生产者生成，也可以由 RabbitMQ 服务器生成。
默认值: 无默认值，可以由生产者生成 UUID 或其他唯一标识符。
代码示例 (Java):


String messageId = java.util.UUID.randomUUID().toString();
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .messageId(messageId)
        .build();
channel.basicPublish("property_exchange", "property.test", properties, message.getBytes("UTF-8"));

10. timestamp (时间戳)

描述: 消息的发送时间，时间戳格式。
用途: 记录消息的发送时间，用于消息延迟分析、消息排序等场景。可以由生产者设置，也可以由 RabbitMQ 服务器自动添加。
默认值: 无默认值，可以由生产者设置当前时间戳。
代码示例 (Python - pika):


import time
properties = pika.BasicProperties(timestamp=int(time.time())) # 设置当前时间戳
channel.basic_publish(exchange='property_exchange',
                      routing_key='property.test',
                      body='Message with timestamp',
                      properties=properties)

11. type (消息类型)

描述: 消息类型的字符串描述。
用途: 用于标识消息的业务类型，方便消费者根据消息类型进行不同的处理。例如，可以区分 "订单创建" 消息和 "支付完成" 消息。
默认值: 无默认值
代码示例 (Java):


AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .type("order.created") // 设置消息类型为 order.created
        .build();
channel.basicPublish("property_exchange", "property.test", properties, message.getBytes("UTF-8"));

12. user_id (用户 ID)

描述: 发送消息的用户 ID。
用途: 用于安全审计和访问控制。需要注意的是，RabbitMQ 会验证连接的用户是否与消息属性中设置的 user_id 一致，如果不一致，消息会被拒绝发布。
默认值: 无默认值，通常由 RabbitMQ 服务器根据连接信息自动设置。
代码示例 (Python - pika，需要配置 RabbitMQ 用户权限):


# 注意： 设置 user_id 需要 RabbitMQ 服务器配置相应的用户和权限
properties = pika.BasicProperties(user_id='producer_user')
channel.basic_publish(exchange='property_exchange',
                      routing_key='property.test',
                      body='Message with user_id',
                      properties=properties)

13. app_id (应用 ID)

描述: 发送消息的应用 ID。
用途: 用于标识发送消息的应用，方便监控和管理。
默认值: 无默认值
代码示例 (Java):


AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .appId("order-service") // 设置应用 ID 为 order-service
        .build();
channel.basicPublish("property_exchange", "property.test", properties, message.getBytes("UTF-8"));

14. cluster_id (集群 ID)

描述: RabbitMQ 集群的 ID。
用途: 在集群环境中，用于标识消息来源的集群。通常由 RabbitMQ 服务器自动设置。
默认值: 无默认值，通常由 RabbitMQ 服务器自动设置。

3.9.2 消息属性的代码实践与最佳实践

代码实践总结:

设置消息属性: 在发布消息时，通过构建 BasicProperties 对象 (或类似结构) 来设置消息属性，并将其作为参数传递给 basic_publish 方法。
获取消息属性: 在消费者端，通过 delivery.getProperties() 方法 (或类似方法) 获取 BasicProperties 对象，然后可以访问各个属性。
不同客户端语言: 不同客户端语言的 API 略有差异，但核心概念和属性名称是相同的。参考对应客户端的官方文档和示例代码。

最佳实践:

按需设置属性: 只设置必要的属性，避免过度使用属性导致消息体积增大和处理复杂性增加。
合理使用 headers: headers 提供了强大的扩展性，但也要避免滥用，保持 headers 的简洁和清晰。
持久化重要消息: 对于重要的业务消息，务必设置 delivery_mode 为 2 (Persistent)，确保消息的可靠性。
使用 correlation_id 和 reply_to 实现请求-响应模式: 在需要请求-响应交互的场景中，合理使用 correlation_id 和 reply_to 可以简化消息处理逻辑。
设置 expiration 防止消息堆积: 对于有时效性的消息，设置 expiration 可以防止消息在队列中长期堆积，浪费资源。
利用 content_type 和 content_encoding 实现协议兼容性: 确保不同系统之间能够正确解析和处理消息。
考虑安全性: 如果消息属性中包含敏感信息，需要考虑加密和访问控制等安全措施。

3.9.3 Mermaid 图解消息属性的应用

以下使用 Mermaid graph TD 图示展示消息属性在消息路由和消息处理中的应用：

图示说明:

Producer: 消息生产者发布消息，消息中包含各种属性 (Message Properties)。
Exchange: Exchange 接收消息，并根据路由键 (Routing Key) 和消息属性 (Message Properties) 进行路由决策 (Routing Decision)。例如，Header Exchange 可以根据 headers 属性进行路由。
Queue 1 & Queue 2: 消息根据路由规则被路由到不同的队列 (Queue 1, Queue 2)。
Consumer 1 & Consumer 2: 不同的消费者 (Consumer 1, Consumer 2) 从各自的队列中消费消息。
Message Properties 子图: 详细展示了消息属性及其用途：
- content_type 和 headers 用于路由决策 (Routing Decision)。
- delivery_mode 影响消息的持久性 (Persistence)。
- expiration 设置消息的过期时间 (Message TTL)。
- priority 用于优先级队列 (Priority Queue)。
- correlation_id 和 reply_to 支持请求-响应模式 (Request-Response)。

此图示清晰地展现了消息属性在 RabbitMQ 消息流转和处理过程中的关键作用。

结论

RabbitMQ 消息属性是消息特性的重要组成部分，它们为消息赋予了丰富的元数据和灵活的处理方式。掌握消息属性的用法，可以帮助开发者构建更可靠、更智能、更高效的消息驱动系统。通过本文的详细解析和代码实践，相信读者已经对 RabbitMQ 消息属性有了更深入的理解，并能够在实际项目中灵活应用，提升系统的整体性能和可靠性。在未来的消息队列应用开发中，合理利用消息属性，将能够更好地满足各种复杂业务场景的需求，构建更加健壮和可扩展的分布式系统。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/216-rabbitmq%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97%E5%AE%9E%E6%88%98%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流