拦截消息

概述

RabbitMQ 提供了一种通用的机制来拦截代理服务器上的消息。拦截可以在两个阶段进行：

1. 传入消息 (Incoming messages) – 在消息进入 RabbitMQ 后，被路由到 队列 之前进行拦截。
2. 传出消息 (Outgoing messages) – 在 RabbitMQ 向客户端发送消息，且在消息被 转换 为目标协议之前进行拦截。

拦截器由以下 Erlang 进程 之一执行：

• AMQP 1.0 会话
• AMQP 0.9.1 通道 (channel)
• MQTT 连接

通过 RabbitMQ Streams 协议 发送的消息不会被拦截。

消息拦截器是一个实现了 rabbit_msg_interceptor 行为的 Erlang 模块。拦截器的具体功能完全取决于其实现——它可以验证消息元数据、添加注释或执行任意的副作用。

可以通过 插件 开发和集成自定义拦截器。

RabbitMQ 自带了几个内置的消息拦截器。以下是如何使用 rabbitmq.conf 文件配置它们的示例。

传入消息拦截器

时间戳

此拦截器为每条传入的消息添加一个时间戳。

message_interceptors.incoming.set_header_timestamp.overwrite = true

• AMQP 1.0 和 Streams 客户端会收到一条消息注释：x-opt-rabbitmq-received-time（自 1970 年 1 月 1 日 UTC 起的毫秒级时间戳）。
• AMQP 0.9.1 客户端会收到：
  • timestamp_in_ms 消息头（毫秒），以便与旧版的 消息时间戳插件 兼容。
  • timestamp 属性（秒）。

若要保留现有的 timestamp_in_ms 消息头，请将 overwrite 设置为 false。

message_interceptors.incoming.set_header_timestamp.overwrite = false

路由节点

此拦截器添加一条 x-routed-by 消息注释，指示接收并路由该消息的 RabbitMQ 节点。

message_interceptors.incoming.set_header_routing_node.overwrite = true

将 overwrite 设置为 false 以保留现有值。

message_interceptors.incoming.set_header_routing_node.overwrite = false

MQTT 客户端 ID

如果启用了 MQTT 插件，RabbitMQ 可以通过添加键为 x-opt-mqtt-client-id 的消息注释，用发布消息的 MQTT 客户端的 客户端 ID 来标记传入消息。

mqtt.message_interceptors.incoming.set_client_id_annotation.enabled = true

该注释对 AMQP 1.0、AMQP 0.9.1 和 Streams 消费者可见。但是，MQTT 客户端将不会收到此注释，因为 MQTT 规范不允许代理服务器添加任意注释。

传出消息拦截器

时间戳

此拦截器在消息发送给客户端时为其添加时间戳。

message_interceptors.outgoing.timestamp.enabled = true

注释键为 x-opt-rabbitmq-sent-time，其值为自 1970 年 1 月 1 日 UTC 起的毫秒级时间戳。