rabbitmq - 无法使用 MessageProperties 将 header 设置为 AMQP 消息

Question

下面的代码显示了我如何为 AMQP 消息设置 header 和消息类型。

MessageProperties properties = new MessageProperties();
    properties.setHeader("KEY", "HOUSE");
    properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
Message message = new Message("1234567;Branch A;SALES;3000.50;Pending approval".getBytes(), properties);
rabbitTemplate.sendAndReceive("", QUEUE_NAME, message);

发送队列中的消息后，消息被Transformer接收。

@Transformer(inputChannel = "inboundChannel", outputChannel = "toutboundChannel")
public Property buildProperty(Message<String> property){
    LOGGER.info("message received :: HEADERS: {}, PAYLOAD :{}", property.getHeaders(), property.getPayload());
....
}    

在日志中，标题“KEY: HOUSE”丢失，甚至消息类型也不是 JSON，而是“text/plain”。

LOGS:
[SimpleAsyncTaskExecutor-1] INFO com.demo.maven.spring.integration.endpoint.TransformerRequestBuilder - message received :: HEADERS: {amqp_receivedRoutingKey=mobile.queue, amqp_deliveryTag=2, amqp_replyTo=amq.rabbitmq.reply-to.g2dkABByYWJiaXRAbG9jYWxob3N0AAAW9QAAAAAD.tTIFOS2gsM7qIlGYaybfrg==, amqp_deliveryMode=PERSISTENT, amqp_redelivered=true, id=399dda4f-4ba1-7cf4-2310-03dbfbac82b6, contentType=text/plain, timestamp=1421649922840}, PAYLOAD :1234567;Branch A;SALES;3000.50;Pending approval

Answer 1

1. MessagePropertiesBuilder 类就是为此目的。

2. 默认情况下，Spring Integration AMQP 入站端点(AmqpInboundChannelAdapter 和 AmqpInboundGateway)仅映射标准 AMQP header 。这是 DefaultAmqpHeaderMapper 的默认行为。要接受任何用户特定的 header ，您应该使用选项 setRequestHeaderNames("*") 将 AmqpHeaderMapper (setHeaderMapper) 注入(inject)到该入站端点。或者提供所需自定义 header 名称的完整列表。

3. 回复。 contentType=text/plain:我认为 AMQP Inbound Endpoint 和 @Transformer(inputChannel = "inboundChannel" 会覆盖从 AMQP contentType 接收到的内容 header 。因为 RabbitTemplate 不这样做，如果您发送 Message 而不是任何其他对象。请共享 DEBUG 日志 org.springframework.integration类为消息接收者。当然我们需要那部分日志，当你收到消息直到那个@Transformer