Message

Spring Integration Message 是一个通用数据容器。任何对象都可以作为负载提供，并且每个 Message 实例都包含以键值对形式的用户可扩展属性的头部。

Message 接口

以下列表显示了 Message 接口的定义

public interface Message<T> {

    T getPayload();

    MessageHeaders getHeaders();

}

Message 接口是 API 的核心部分。通过将数据封装在通用包装器中，消息系统可以在不知道数据类型的情况下传递它。随着应用程序发展以支持新类型，或者当类型本身被修改或扩展时，消息系统不会受到影响。另一方面，当消息系统中的某些组件确实需要访问有关 Message 的信息时，此类元数据通常可以存储到消息头部中的元数据并从中检索。

消息头部

正如 Spring Integration 允许将任何 Object 用作 Message 的有效负载一样，它也支持将任何 Object 类型用作头部值。实际上，MessageHeaders 类实现了 java.util.Map_ 接口，如下面的类定义所示

public final class MessageHeaders implements Map<String, Object>, Serializable {
  ...
}
尽管 MessageHeaders 类实现了 Map，但它实际上是一个只读实现。任何尝试在 Map 中 put 值的操作都会导致 UnsupportedOperationException。remove 和 clear 也同样适用。由于消息可能会传递给多个消费者，因此 Map 的结构无法修改。同样，消息的有效负载 Object 在初始创建后不能 set。但是，头部值本身（或有效负载 Object）的可变性有意地留作框架用户的决定。

作为 Map 的实现，可以通过使用头部名称调用 get(..) 来检索头部。或者，您可以提供预期的 Class 作为附加参数。更好的是，当检索预定义值之一时，可以使用方便的 getter。以下示例显示了这三个选项中的每一个

Object someValue = message.getHeaders().get("someKey");

CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class);

Long timestamp = message.getHeaders().getTimestamp();

下表描述了预定义的消息头部

表 1. 预定义的消息头部
头部名称 头部类型 用法 
 MessageHeaders.ID
 
 java.util.UUID

此消息实例的标识符。每次消息被修改时都会改变。

 
 MessageHeaders.
TIMESTAMP
 
 java.lang.Long

消息创建的时间。每次消息被修改时都会改变。

 
 MessageHeaders.
REPLY_CHANNEL
 
 java.lang.Object
(String or
MessageChannel)

当没有配置显式输出通道且没有 ROUTING_SLIP 或 ROUTING_SLIP 已耗尽时，用于发送回复（如果有）的通道。如果值为 String，则它必须表示一个 bean 名称或由 ChannelRegistry 生成。

 
 MessageHeaders.
ERROR_CHANNEL
 
 java.lang.Object
(String or
MessageChannel)

用于发送错误的通道。如果值为 String，则它必须表示一个 bean 名称或由 ChannelRegistry 生成。

许多入站和出站适配器实现也提供或期望某些头部，您可以配置其他用户定义的头部。这些头部的常量可以在存在此类头部的模块中找到，例如 AmqpHeaders、JmsHeaders 等等。

MessageHeaderAccessor API

从 Spring Framework 4.0 和 Spring Integration 4.0 开始，核心消息抽象已移至 spring-messaging 模块，并引入了 MessageHeaderAccessor API 以提供对消息实现的额外抽象。所有（核心）Spring Integration 特定的消息头部常量现在都在 IntegrationMessageHeaderAccessor 类中声明。下表描述了预定义的消息头部

表 2. 预定义的消息头部
头部名称 头部类型 用法 
 IntegrationMessageHeaderAccessor.
CORRELATION_ID
 
 java.lang.Object

用于关联两个或多个消息。

 
 IntegrationMessageHeaderAccessor.
SEQUENCE_NUMBER
 
 java.lang.Integer

通常是具有 SEQUENCE_SIZE 的一组消息中的序列号，但也可以在 <resequencer/> 中用于重新排序无界限的一组消息。

 
 IntegrationMessageHeaderAccessor.
SEQUENCE_SIZE
 
 java.lang.Integer

一组关联消息中的消息数量。

 
 IntegrationMessageHeaderAccessor.
EXPIRATION_DATE
 
 java.lang.Long

指示消息何时过期。框架不直接使用，但可以通过头部丰富器设置，并在配置了 UnexpiredMessageSelector 的 <filter/> 中使用。

 
 IntegrationMessageHeaderAccessor.
PRIORITY
 
 java.lang.Integer

消息优先级——例如，在 PriorityChannel 中。

 
 IntegrationMessageHeaderAccessor.
DUPLICATE_MESSAGE
 
 java.lang.Boolean

如果消息被幂等接收器拦截器检测为重复。请参阅 幂等接收器企业集成模式。

 
 IntegrationMessageHeaderAccessor.
CLOSEABLE_RESOURCE
 
 java.io.Closeable

如果消息与一个 Closeable 关联，则存在此头部，该 Closeable 应在消息处理完成后关闭。一个示例是使用 FTP、SFTP 等进行流式文件传输的 Session。

 
 IntegrationMessageHeaderAccessor.
DELIVERY_ATTEMPT
 
 java.lang.
AtomicInteger

如果消息驱动通道适配器支持配置 RetryTemplate，则此头部包含当前的交付尝试。

 
 IntegrationMessageHeaderAccessor.
ACKNOWLEDGMENT_CALLBACK
 
 o.s.i.support.
Acknowledgment
Callback

如果入站端点支持，则回调以接受、拒绝或重新排队消息。请参阅 延迟确认可轮询消息源 和 MQTT 手动确认。

IntegrationMessageHeaderAccessor 类上提供了其中一些头部的便捷类型化 getter，如下例所示

IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(message);
int sequenceNumber = accessor.getSequenceNumber();
Object correlationId = accessor.getCorrelationId();
...

下表描述了也出现在 IntegrationMessageHeaderAccessor 中但通常不被用户代码使用（即，它们通常被 Spring Integration 内部部分使用——此处包含它们是为了完整性）的头部

表 3. 预定义的消息头部
头部名称 头部类型 用法 
 IntegrationMessageHeaderAccessor.
SEQUENCE_DETAILS
 
 java.util.
List<List<Object>>

当需要嵌套关联时（例如，splitter→…​→splitter→…​→aggregator→…​→aggregator），使用的一组关联数据。

 
 IntegrationMessageHeaderAccessor.
ROUTING_SLIP
 
 java.util.
Map<List<Object>, Integer>

请参阅 路由单。

消息 ID 生成

当消息通过应用程序转换时，每次修改它时（例如，通过转换器），都会分配一个新的消息 ID。消息 ID 是一个 UUID。从 Spring Integration 3.0 开始，用于 IS 生成的默认策略比以前的 java.util.UUID.randomUUID() 实现更有效。它使用基于安全随机种子的简单随机数，而不是每次都创建一个安全随机数。

可以通过在应用程序上下文中声明一个实现 org.springframework.util.IdGenerator 的 bean 来选择不同的 UUID 生成策略。

在一个类加载器中只能使用一个 UUID 生成策略。这意味着如果两个或多个应用程序上下文在同一个类加载器中运行，它们将共享相同的策略。如果其中一个上下文更改了策略，则所有上下文都将使用该策略。如果同一个类加载器中的两个或多个上下文声明了类型为 org.springframework.util.IdGenerator 的 bean，则它们都必须是同一类的实例。否则，尝试替换自定义策略的上下文将无法初始化。如果策略相同，但参数化，则使用第一个初始化的上下文中的策略。

除了默认策略之外，还提供了另外两个 IdGenerators。org.springframework.util.JdkIdGenerator 使用之前的 UUID.randomUUID() 机制。当不需要 UUID 并且简单的递增值就足够时，您可以使用 o.s.i.support.IdGenerators.SimpleIncrementingIdGenerator。

只读头部

MessageHeaders.ID 和 MessageHeaders.TIMESTAMP 是只读头部，不能被覆盖。

从 4.3.2 版本开始，MessageBuilder 提供了 readOnlyHeaders(String…​ readOnlyHeaders) API，用于自定义不应从上游 Message 复制的头部列表。默认情况下，只有 MessageHeaders.ID 和 MessageHeaders.TIMESTAMP 是只读的。提供了全局 spring.integration.readOnly.headers 属性（参见 全局属性）来为框架组件自定义 DefaultMessageBuilderFactory。当您不想由 ObjectToJsonTransformer（参见 JSON 转换器）填充一些开箱即用的头部（例如 contentType）时，这会很有用。

当您尝试使用 MessageBuilder 构建新消息时，此类头部将被忽略，并且会在日志中发出特定的 INFO 消息。

从 5.0 版本开始，消息网关、头部丰富器、内容丰富器 和 头部过滤器 在使用 DefaultMessageBuilderFactory 时不允许您配置 MessageHeaders.ID 和 MessageHeaders.TIMESTAMP 头部名称，并且会抛出 BeanInitializationException。

头部传播

当消息生产者端点（例如 服务激活器）处理（和修改）消息时，通常，入站头部会传播到出站消息。一个例外是 转换器，当一个完整的消息返回到框架时。在这种情况下，用户代码负责整个出站消息。当转换器只返回有效负载时，入站头部会传播。此外，只有当头部在出站消息中尚不存在时才会传播，从而允许您根据需要更改头部值。

从 4.3.10 版本开始，您可以配置消息处理器（修改消息并产生输出）以抑制特定头部的传播。要配置您不想复制的头部，请在 MessageProducingMessageHandler 抽象类上调用 setNotPropagatedHeaders() 或 addNotPropagatedHeaders() 方法。

您还可以通过将 META-INF/spring.integration.properties 中的 readOnlyHeaders 属性设置为逗号分隔的头部列表来全局抑制特定消息头部的传播。

从 5.0 版本开始，AbstractMessageProducingHandler 上的 setNotPropagatedHeaders() 实现应用简单模式（xxx*、*xxx、*xxx* 或 xxx*yyy）以允许使用通用后缀或前缀过滤头部。有关更多信息，请参阅 PatternMatchUtils Javadoc。当其中一个模式是 *（星号）时，不传播任何头部。所有其他模式都将被忽略。在这种情况下，服务激活器与转换器的行为方式相同，并且任何所需的头部都必须在服务方法返回的 Message 中提供。notPropagatedHeaders() 选项在 Java DSL 的 ConsumerEndpointSpec 中可用。它也可用于 <service-activator> 组件的 XML 配置，作为 not-propagated-headers 属性。

头部传播抑制不适用于不修改消息的端点，例如 桥接器 和 路由器。

消息实现

Message 接口的基本实现是 GenericMessage<T>，它提供两个构造函数，如下所示

new GenericMessage<T>(T payload);

new GenericMessage<T>(T payload, Map<String, Object> headers)

当创建 Message 时，会生成一个随机的唯一 ID。接受 Map 头部的构造函数会将提供的头部复制到新创建的 Message 中。

还有一个方便的 Message 实现，旨在传达错误情况。此实现以 Throwable 对象作为其有效负载，如下例所示

ErrorMessage message = new ErrorMessage(someThrowable);

Throwable t = message.getPayload();

请注意，此实现利用了 GenericMessage 基类参数化的事实。因此，如两个示例所示，在检索 Message 有效负载 Object 时不需要进行任何强制转换。

上述 Message 类实现是不可变的。在某些情况下，当可变性不是问题并且应用程序逻辑设计良好以避免并发修改时，可以使用 MutableMessage。

MessageBuilder 辅助类

您可能会注意到 Message 接口定义了其有效负载和头部的检索方法，但没有提供 setter。其原因是 Message 在初始创建后不能被修改。因此，当 Message 实例发送给多个消费者时（例如，通过发布/订阅通道），如果其中一个消费者需要发送带有不同有效负载类型的回复，它必须创建一个新的 Message。这样，其他消费者就不会受到这些更改的影响。请记住，多个消费者可能访问相同的有效负载实例或头部值，并且此类实例本身是否不可变由您决定。换句话说，Message 实例的契约类似于不可修改的 Collection，MessageHeaders 映射进一步证明了这一点。尽管 MessageHeaders 类实现了 java.util.Map，但任何尝试对 MessageHeaders 实例调用 put 操作（或“remove”或“clear”）都会导致 UnsupportedOperationException。

Spring Integration 提供了一种更方便的方法来构建消息，而不是要求创建和填充 Map 以传递给 GenericMessage 构造函数：MessageBuilder。MessageBuilder 提供两种工厂方法，用于从现有 Message 或使用有效负载 Object 创建 Message 实例。当从现有 Message 构建时，该 Message 的头部和有效负载将复制到新的 Message 中，如下例所示

Message<String> message1 = MessageBuilder.withPayload("test")
        .setHeader("foo", "bar")
        .build();

Message<String> message2 = MessageBuilder.fromMessage(message1).build();

assertEquals("test", message2.getPayload());
assertEquals("bar", message2.getHeaders().get("foo"));

如果您需要创建具有新有效负载但仍希望从现有 Message 复制头部的 Message，您可以使用其中一个“copy”方法，如下例所示

Message<String> message3 = MessageBuilder.withPayload("test3")
        .copyHeaders(message1.getHeaders())
        .build();

Message<String> message4 = MessageBuilder.withPayload("test4")
        .setHeader("foo", 123)
        .copyHeadersIfAbsent(message1.getHeaders())
        .build();

assertEquals("bar", message3.getHeaders().get("foo"));
assertEquals(123, message4.getHeaders().get("foo"));

请注意，copyHeadersIfAbsent 方法不会覆盖现有值。此外，在上面的示例中，您可以看到如何使用 setHeader 设置任何用户定义的头部。最后，预定义头部也提供了 set 方法，以及一个用于设置任何头部（MessageHeaders 也定义了预定义头部名称的常量）的非破坏性方法。

您还可以使用 MessageBuilder 设置消息的优先级，如下例所示

Message<Integer> importantMessage = MessageBuilder.withPayload(99)
        .setPriority(5)
        .build();

assertEquals(5, importantMessage.getHeaders().getPriority());

Message<Integer> lessImportantMessage = MessageBuilder.fromMessage(importantMessage)
        .setHeaderIfAbsent(IntegrationMessageHeaderAccessor.PRIORITY, 2)
        .build();

assertEquals(2, lessImportantMessage.getHeaders().getPriority());

priority 头部仅在使用 PriorityChannel 时才会被考虑（如下一章所述）。它被定义为 java.lang.Integer。

提供了 MutableMessageBuilder 来处理 MutableMessage 实例。此类的逻辑是创建 MutableMessage 或保持原样，并通过构建器方法修改其内容。这样，当不可变性不是消息交换的关注点时，运行中的应用程序的性能会略有提高。

从版本 6.4 开始，BaseMessageBuilder 类从 MessageBuilder 中提取出来，以简化对默认消息构建逻辑的扩展。例如，结合自定义 MessageBuilderFactory，自定义 BaseMessageBuilder 实现可以在应用程序上下文中全局使用，以提供自定义 Message 实例。特别是，可以覆盖 GenericMessage.toString() 方法，以便在记录此类消息时隐藏有效负载和头部中的敏感信息。

MessageBuilderFactory 抽象

名为 IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME 的 MessageBuilderFactory bean 被全局注册到应用程序上下文中，并在框架的任何地方用于创建 Message 实例。默认情况下，它是一个 DefaultMessageBuilderFactory 实例。开箱即用，框架还提供了一个 MutableMessageBuilderFactory，用于在框架组件中创建 MutableMessage 实例。为了自定义 Message 实例的创建，必须在目标应用程序上下文中提供一个名为 IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME 的 MessageBuilderFactory bean，以覆盖默认的。例如，可以为 BaseMessageBuilder 的实现注册一个自定义 MessageBuilderFactory，在该实现中，我们希望提供一个扩展了 GenericMessage 并覆盖了 toString() 的实例，以便在记录此类消息时隐藏有效负载和头部中的敏感信息。

这些类的快速实现可以像这样演示个人身份信息缓解

class PiiMessageBuilderFactory implements MessageBuilderFactory {

	@Override
	public <T> PiiMessageBuilder<T> fromMessage(Message<T> message) {
	    return new PiiMessageBuilder<>(message.getPayload(), message);
	}

	@Override
	public <T> PiiMessageBuilder<T> withPayload(T payload) {
	    return new PiiMessageBuilder<>(payload, null);
	}

}

class PiiMessageBuilder<P> extends BaseMessageBuilder<P, PiiMessageBuilder<P>> {

    public PiiMessageBuilder(P payload, @Nullable Message<P> originalMessage) {
        super(payload, originalMessage);
    }

    @Override
    public Message<P> build() {
        return new PiiMessage<>(getPayload(), getHeaders());
    }

}

class PiiMessage<P> extends GenericMessage<P> {

    @Serial
    private static final long serialVersionUID = -354503673433669578L;

    public PiiMessage(P payload, Map<String, Object> headers) {
        super(payload, headers);
    }

    @Override
    public String toString() {
        return "PiiMessage [payload=" + getPayload() + ", headers=" + maskHeaders(getHeaders()) + ']';
    }

    private static Map<String, Object> maskHeaders(Map<String, Object> headers) {
        return headers.entrySet()
                .stream()
                .map((entry) -> entry.getKey().equals("password") ? Map.entry(entry.getKey(), "******") : entry)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

}

然后可以将此 PiiMessageBuilderFactory 注册为 bean，每当框架记录消息时（例如在 errorChannel 的情况下），password 头部将被屏蔽。

© . This site is unofficial and not affiliated with VMware.