AMQP 消息头
概述
Spring Integration AMQP 适配器会自动映射所有 AMQP 属性和消息头。(这是从 4.3 版本开始的变更 - 以前仅映射标准消息头)。默认情况下,这些属性使用 DefaultAmqpHeaderMapper
在 Spring Integration MessageHeaders
之间来回复制。
您可以传入自己的 AMQP 特定消息头映射器实现,因为适配器支持通过属性进行此操作。
AMQP MessageProperties
中的任何用户定义消息头都会复制到 AMQP 消息或从 AMQP 消息复制,除非被 DefaultAmqpHeaderMapper
的 requestHeaderNames
或 replyHeaderNames
属性明确否定。默认情况下,对于出站映射器,不会映射任何 x-*
消息头。有关原因,请参阅本节后面出现的注意事项。
要覆盖默认设置并恢复到 4.3 版本之前的行为,请在属性中使用 STANDARD_REQUEST_HEADERS
和 STANDARD_REPLY_HEADERS
。
映射用户定义消息头时,值也可以包含简单的通配符模式(例如 thing* 或 *thing )进行匹配。* 匹配所有消息头。 |
从 4.1 版本开始,AbstractHeaderMapper
(DefaultAmqpHeaderMapper
的超类)允许为 requestHeaderNames
和 replyHeaderNames
属性配置 NON_STANDARD_HEADERS
令牌(除了现有的 STANDARD_REQUEST_HEADERS
和 STANDARD_REPLY_HEADERS
),以映射所有用户定义的头。
org.springframework.amqp.support.AmqpHeaders
类标识了 DefaultAmqpHeaderMapper
使用的默认消息头
-
amqp_appId
-
amqp_clusterId
-
amqp_contentEncoding
-
amqp_contentLength
-
content-type
(参见contentType
消息头) -
amqp_correlationId
-
amqp_delay
-
amqp_deliveryMode
-
amqp_deliveryTag
-
amqp_expiration
-
amqp_messageCount
-
amqp_messageId
-
amqp_receivedDelay
-
amqp_receivedDeliveryMode
-
amqp_receivedExchange
-
amqp_receivedRoutingKey
-
amqp_redelivered
-
amqp_replyTo
-
amqp_timestamp
-
amqp_type
-
amqp_userId
-
amqp_publishConfirm
-
amqp_publishConfirmNackCause
-
amqp_returnReplyCode
-
amqp_returnReplyText
-
amqp_returnExchange
-
amqp_returnRoutingKey
-
amqp_channel
-
amqp_consumerTag
-
amqp_consumerQueue
如本节前面所述,使用 * 消息头映射模式是复制所有消息头的常用方法。然而,这可能会产生一些意外的副作用,因为某些 RabbitMQ 专有属性/消息头也会被复制。例如,当您使用联合交换器 (federation) 时,接收到的消息可能有一个名为 x-received-from 的属性,其中包含发送消息的节点。如果您在入站网关上使用通配符 * 进行请求和回复消息头映射,此消息头会被复制,这可能会导致联合出现问题。此回复消息可能会被联合回发送代理,发送代理可能会认为消息正在循环,并因此静默丢弃它。如果您希望使用通配符消息头映射的便利性,您可能需要在下游流中过滤掉一些消息头。例如,为了避免将 x-received-from 消息头复制回回复,您可以在将回复发送到 AMQP 入站网关之前使用 <int:header-filter … header-names="x-received-from"> 。或者,您可以明确列出您实际希望映射的属性,而不是使用通配符。出于这些原因,对于入站消息,映射器(默认情况下)不映射任何 x-* 消息头。它也不将 deliveryMode 映射到 amqp_deliveryMode 消息头,以避免该消息头从入站消息传播到出站消息。相反,此消息头被映射到 amqp_receivedDeliveryMode ,该消息头在输出时不会被映射。 |
从 4.3 版本开始,消息头映射中的模式可以通过在模式前加上 !
来否定。被否定的模式具有优先级,因此像 STANDARD_REQUEST_HEADERS,thing1,ba*,!thing2,!thing3,qux,!thing1
这样的列表不会映射 thing1
(也不会映射 thing2
和 thing3
)。标准消息头以及 bad
和 qux
会被映射。否定技术对于例如在接收器下游以不同方式执行 JSON 反序列化逻辑时不映射 JSON 类型消息头非常有用。为此,应为入站通道适配器/网关的消息头映射器配置 !json_*
模式。
如果您有一个以 ! 开头的用户定义消息头,并且您确实希望映射它,则需要使用 \ 进行转义,如下所示:STANDARD_REQUEST_HEADERS,\!myBangHeader 。名为 !myBangHeader 的消息头现在被映射了。 |
从 5.1 版本开始,如果出站消息上不存在相应的 amqp_messageId 或 amqp_timestamp 消息头,DefaultAmqpHeaderMapper 将回退到分别将 MessageHeaders.ID 和 MessageHeaders.TIMESTAMP 映射到 MessageProperties.messageId 和 MessageProperties.timestamp 。入站属性将像以前一样映射到 amqp_* 消息头。在消息消费者使用有状态重试时,填充 messageId 属性非常有用。 |
contentType
消息头
与其他消息头不同,AmqpHeaders.CONTENT_TYPE
没有前缀 amqp_
;这使得 contentType 消息头可以在不同的技术之间透明传递。例如,发送到 RabbitMQ 队列的入站 HTTP 消息。
contentType
消息头被映射到 Spring AMQP 的 MessageProperties.contentType
属性,然后该属性被映射到 RabbitMQ 的 content_type
属性。
在 5.1 版本之前,此消息头也被映射为 MessageProperties.headers
map 中的一个条目;这是不正确的,而且,由于底层的 Spring AMQP 消息转换器可能改变了内容类型,该值可能是错误的。这种变化会反映在一等(first-class)的 content_type
属性中,但不会反映在 RabbitMQ headers map 中。入站映射会忽略 headers map 中的值。contentType
不再映射到 headers map 中的一个条目。