在 org.springframework.amqp.core.Message class 中使用 HttpHeader

Question

我正在尝试将传入的 HttpHeader 保存在 org.springframework.amqp.core.Message 中，以便稍后从 queue 和 cleanly 中提取消息，并在 RestTemplate 中用作 HttpHeader。它对我来说似乎不太干净。

我在 spring 休息控制器中有一个 @PostMapping，它采用这样的入站 HttpHeader：

@PostMapping
public ResponseEntity<String> inbound(@PathVariable String flow, @RequestHeader HttpHeaders headers, @RequestBody byte[] payload) {

然后在 org.springframework.amqp.core.Message 中调用 createMessage（下方）方法以将 HttpHeader 存储在 Message 中。

像我这样从 org.springframework.amqp.core.Message 中检索 headers 似乎不太干净，因为：

我假设每个值都是一个字符串，但事实可能并非如此
有时我必须删除“[”和“]”，因为列表 object 可能会从 Map<String, Object> headers = message.getMessageProperties.getHeaders();

有谁知道将 headers 从 Message 中拉出并将它们添加到 HttpHeaders object 的更简洁的方法？我所拥有的似乎相当假设和笨拙，因为我不想：

假设每个值都是一个字符串（就像我在下面做的那样）
使用 instanceof 并分别对每种数据类型进行类型转换？

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.http.HttpHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

@Component
public class MessageHelper {

    public Message createMessage(HttpHeaders httpHeaders, byte[] payload) {
        Map<String, Object> headers = new HashMap<>();
        headers.putAll(httpHeaders);

        return MessageBuilder
                .withBody(payload)
                .copyHeaders(new MessageHeaders(headers))
                .setContentType(httpHeaders.getContentType().toString())
                .setContentEncoding(StandardCharsets.UTF_8.displayName()).build();
    }

    public HttpHeaders getHeaders(Message message) {
        HttpHeaders httpHeaders = new HttpHeaders();
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        for (Map.Entry<String, Object> header : headers.entrySet()) {
            String value = header.getValue().toString().replace("]", "").replace("[", "");
            httpHeaders.addIfAbsent(header.getKey(), value);
        }

        return httpHeaders;
    }
}

注意：我没有使用 spring-integrations，它似乎有其他选项，例如 MessageHeaderAccessor 或 spring-messaging，它似乎有 MessageHeaders - 但我会遗漏什么？

Answer 1

请参阅此处 https://github.com/spring-projects/spring-amqp/blob/main/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java#L176-L223，了解 Spring 在写入 AMQP 时如何映射任意 header。

如果 header 值不是 AMQP 支持的类型之一，则会对其执行简单的 toString() 操作（因此 [...]，如果它类似于Set).

支持的类型：

        boolean valid = (value instanceof String) || (value instanceof byte[])
                || (value instanceof Boolean) || (value instanceof Class)
                || (value instanceof LongString) || (value instanceof Integer) || (value instanceof Long)
                || (value instanceof Float) || (value instanceof Double) || (value instanceof BigDecimal)
                || (value instanceof Short) || (value instanceof Byte) || (value instanceof Date)
                || (value instanceof List) || (value instanceof Map) || (value instanceof Object[]);

对于容器类型（数组、列表），递归调用该方法。

一个解决方案可能是将 header 序列化为单个 byte[] header.

在 org.springframework.amqp.core.Message class 中使用 HttpHeader

Using HttpHeaders in org.springframework.amqp.core.Message class

amqp

rabbitmq

spring-rabbit

spring-amqp

spring-boot