STOMP 支持

Spring 集成 4.2 版本引入了 STOMP（简单文本面向消息协议）客户端支持。它基于 Spring 框架消息模块、stomp 包的架构、基础设施和 API。Spring 集成使用许多 Spring STOMP 组件（例如 StompSession 和 StompClientSupport）。有关更多信息，请参阅 Spring 框架参考手册中的 Spring 框架 STOMP 支持章节。

您需要将此依赖项包含到您的项目中

Maven
Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stomp</artifactId>
    <version>6.3.0</version>
</dependency>

compile "org.springframework.integration:spring-integration-stomp:6.3.0"

对于服务器端组件，您需要添加 org.springframework:spring-websocket 和/或 io.projectreactor.netty:reactor-netty 依赖项。

概述

要配置 STOMP，您应该从 STOMP 客户端对象开始。Spring 框架提供了以下实现

WebSocketStompClient：基于 Spring WebSocket API，支持标准 JSR-356 WebSocket、Jetty 9 和 SockJS，用于使用 SockJS 客户端进行基于 HTTP 的 WebSocket 模拟。
ReactorNettyTcpStompClient：基于 reactor-netty 项目中的 ReactorNettyTcpClient。

您可以提供任何其他 StompClientSupport 实现。请参阅这些类的 Javadoc。

StompClientSupport 类被设计为一个工厂，用于为提供的 StompSessionHandler 生成 StompSession，所有剩余的工作都通过对该 StompSessionHandler 和 StompSession 抽象的回调来完成。使用 Spring 集成适配器抽象，我们需要提供一些托管的共享对象来表示我们的应用程序作为具有其唯一会话的 STOMP 客户端。为此，Spring 集成提供了 StompSessionManager 抽象来管理任何提供的 StompSessionHandler 之间的单个 StompSession。这允许对特定 STOMP 代理使用入站或出站通道适配器（或两者）。有关更多信息，请参阅 StompSessionManager（及其实现）JavaDocs。

STOMP 入站通道适配器

StompInboundChannelAdapter 是一个一站式 MessageProducer 组件，它将您的 Spring 集成应用程序订阅到提供的 STOMP 目的地，并从这些目的地接收消息（通过在连接的 StompSession 上使用提供的 MessageConverter 从 STOMP 帧转换）。您可以通过在 StompInboundChannelAdapter 上使用适当的 @ManagedOperation 注释来在运行时更改目的地（以及 STOMP 订阅）。

有关更多配置选项，请参阅 STOMP 命名空间支持和 StompInboundChannelAdapter 的 Javadoc。

STOMP 出站通道适配器

StompMessageHandler 是 <int-stomp:outbound-channel-adapter> 的 MessageHandler，用于通过 StompSession（由共享的 StompSessionManager 提供）将传出的 Message<?> 实例发送到 STOMP destination（预先配置或在运行时使用 SpEL 表达式确定）。

有关更多配置选项，请参阅 STOMP 命名空间支持和 StompMessageHandler 的 Javadoc。

STOMP 头部映射

STOMP 协议在帧中提供头部。STOMP 帧的整个结构具有以下格式

....
COMMAND
header1:value1
header2:value2

Body^@
....

Spring 框架提供 StompHeaders 来表示这些头部。有关更多详细信息，请参阅 Javadoc。STOMP 帧被转换为 Message<?> 实例，反之亦然，这些头部被映射到 MessageHeaders 实例，反之亦然。Spring Integration 为 STOMP 适配器提供了一个默认的 HeaderMapper 实现。该实现是 StompHeaderMapper。它分别为入站和出站适配器提供 fromHeaders() 和 toHeaders() 操作。

与许多其他 Spring Integration 模块一样，IntegrationStompHeaders 类已被引入，用于将标准 STOMP 头部映射到 MessageHeaders，其中 stomp_ 作为头部名称前缀。此外，所有具有该前缀的 MessageHeaders 实例在发送到目标时都会映射到 StompHeaders。

有关更多信息，请参阅这些类的 Javadoc 以及 STOMP 命名空间支持中的 mapped-headers 属性描述。

STOMP 集成事件

许多 STOMP 操作是异步的，包括错误处理。例如，STOMP 具有一个 RECEIPT 服务器帧，它在客户端帧通过添加 RECEIPT 头部请求一个时返回。为了提供对这些异步事件的访问，Spring Integration 发出 StompIntegrationEvent 实例，您可以通过实现 ApplicationListener 或使用 <int-event:inbound-channel-adapter> 来获取这些实例（请参阅接收 Spring 应用程序事件）。

具体来说，当一个 stompSessionListenableFuture 由于无法连接到 STOMP 代理而收到 onFailure() 时，AbstractStompSessionManager 会发出一个 StompExceptionEvent。另一个例子是 StompMessageHandler。它处理 ERROR STOMP 帧，这些帧是服务器对由该 StompMessageHandler 发送的无效（未接受）消息的响应。

StompMessageHandler 在对发送到 StompSession 的消息的异步回复中，作为 StompSession.Receiptable 回调的一部分，发出 StompReceiptEvent。StompReceiptEvent 可以是正面的或负面的，具体取决于是否在 receiptTimeLimit 时间段内从服务器接收到 RECEIPT 帧，您可以在 StompClientSupport 实例上配置此时间段。默认值为 15 * 1000（以毫秒为单位，即 15 秒）。

只有当要发送的消息的 RECEIPT STOMP 标头不为 null 时，才会添加 StompSession.Receiptable 回调。您可以通过 StompSession 的 autoReceipt 选项和 StompSessionManager 分别启用自动 RECEIPT 标头生成。

有关如何配置 Spring Integration 以接受这些 ApplicationEvent 实例的更多信息，请参见 STOMP 适配器 Java 配置。

STOMP 适配器 Java 配置

以下示例展示了 STOMP 适配器的综合 Java 配置

@Configuration
@EnableIntegration
public class StompConfiguration {

    @Bean
    public ReactorNettyTcpStompClient stompClient() {
        ReactorNettyTcpStompClient stompClient = new ReactorNettyTcpStompClient("127.0.0.1", 61613);
        stompClient.setMessageConverter(new PassThruMessageConverter());
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.afterPropertiesSet();
        stompClient.setTaskScheduler(taskScheduler);
        stompClient.setReceiptTimeLimit(5000);
        return stompClient;
    }

    @Bean
    public StompSessionManager stompSessionManager() {
        ReactorNettyTcpStompSessionManager stompSessionManager = new ReactorNettyTcpStompSessionManager(stompClient());
        stompSessionManager.setAutoReceipt(true);
        return stompSessionManager;
    }

    @Bean
    public PollableChannel stompInputChannel() {
        return new QueueChannel();
    }

    @Bean
    public StompInboundChannelAdapter stompInboundChannelAdapter() {
        StompInboundChannelAdapter adapter =
        		new StompInboundChannelAdapter(stompSessionManager(), "/topic/myTopic");
        adapter.setOutputChannel(stompInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "stompOutputChannel")
    public MessageHandler stompMessageHandler() {
        StompMessageHandler handler = new StompMessageHandler(stompSessionManager());
        handler.setDestination("/topic/myTopic");
        return handler;
    }

    @Bean
    public PollableChannel stompEvents() {
        return new QueueChannel();
    }

    @Bean
    public ApplicationListener<ApplicationEvent> stompEventListener() {
        ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
        producer.setEventTypes(StompIntegrationEvent.class);
        producer.setOutputChannel(stompEvents());
        return producer;
    }

}

STOMP 命名空间支持

Spring Integration STOMP 命名空间实现了入站和出站通道适配器组件。要在您的配置中包含它，请在您的应用程序上下文配置文件中提供以下命名空间声明

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:int="http://www.springframework.org/schema/integration"
  xmlns:int-stomp="http://www.springframework.org/schema/integration/stomp"
  xsi:schemaLocation="
    http://www.springframework.org/schema/beans
    https://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/integration
    https://www.springframework.org/schema/integration/spring-integration.xsd
    http://www.springframework.org/schema/integration/stomp
    https://www.springframework.org/schema/integration/stomp/spring-integration-stomp.xsd">
    ...
</beans>

了解 `<int-stomp:outbound-channel-adapter>` 元素

以下列表显示了 STOMP 出站通道适配器可用的属性

<int-stomp:outbound-channel-adapter
                           id=""                      (1)
                           channel=""                 (2)
                           stomp-session-manager=""   (3)
                           header-mapper=""           (4)
                           mapped-headers=""          (5)
                           destination=""             (6)
                           destination-expression=""  (7)
                           auto-startup=""            (8)
                           phase=""/>                 (9)

1	组件 bean 名称。`MessageHandler` 注册了一个 bean 别名，即 `id` 加上 `.handler`。如果您没有设置 `channel` 属性，则会创建一个 `DirectChannel` 并使用此 `id` 属性的值作为 bean 名称注册到应用程序上下文。在这种情况下，端点注册了一个 bean 名称，即 `id` 加上 `.adapter`。
2	如果存在 `id`，则标识与该适配器关联的通道。请参见 `id`。可选。
3	对 `StompSessionManager` bean 的引用，该 bean 封装了低级连接和 `StompSession` 处理操作。必需。
4	对实现 `HeaderMapper<StompHeaders>` 的 bean 的引用，该 bean 将 Spring Integration `MessageHeaders` 映射到 STOMP 帧头和从 STOMP 帧头映射。它与 `mapped-headers` 互斥。它默认为 `StompHeaderMapper`。
5	要映射到 STOMP 帧头的 STOMP 头名称的逗号分隔列表。只有在未设置 `header-mapper` 引用时才能提供它。此列表中的值也可以是与头名称匹配的简单模式（例如 `myheader` 或 `myheader`）。特殊标记 (`STOMP_OUTBOUND_HEADERS`) 代表所有标准 STOMP 头（内容长度、收据、心跳等）。它们默认包含在内。如果您想添加自己的头并且希望标准头也映射，则必须包含此标记或使用 `header-mapper` 提供您自己的 `HeaderMapper` 实现。
6	发送 STOMP 消息的目标名称。它与 `destination-expression` 互斥。
7	在运行时针对每个 Spring Integration `Message` 作为根对象进行评估的 SpEL 表达式。它与 `destination` 互斥。
8	布尔值，指示此端点是否应自动启动。它默认为 `true`。
9	此端点应启动和停止的生命周期阶段。值越低，此端点启动越早，停止越晚。默认值为 `Integer.MIN_VALUE`。值可以为负数。请参见 `SmartLifeCycle`.

了解 `<int-stomp:inbound-channel-adapter>` 元素

以下列表显示了 STOMP 入站通道适配器可用的属性

<int-stomp:inbound-channel-adapter
                           id=""                     (1)
                           channel=""                (2)
                           error-channel=""          (3)
                           stomp-session-manager=""  (4)
                           header-mapper=""          (5)
                           mapped-headers=""         (6)
                           destinations=""           (7)
                           send-timeout=""           (8)
                           payload-type=""           (9)
                           auto-startup=""           (10)
                           phase=""/>                (11)

1	组件 bean 名称。如果您未设置 `channel` 属性，则会创建一个 `DirectChannel` 并使用此 `id` 属性的值作为 bean 名称在应用程序上下文中注册。在这种情况下，端点使用 bean 名称 `id` 加上 `.adapter` 注册。
2	标识与该适配器关联的通道。
3	发送到`ErrorMessage`实例的`MessageChannel` bean 引用。
4	参见`<int-stomp:outbound-channel-adapter>`上的相同选项。
5	从 STOMP 帧头映射的 STOMP 头部的名称的逗号分隔列表。只有在未设置`header-mapper`引用时才能提供此选项。此列表中的值也可以是与头名匹配的简单模式（例如，`myheader`或`myheader`）。特殊标记（`STOMP_INBOUND_HEADERS`）表示所有标准 STOMP 头部（内容长度、回执、心跳等）。它们默认包含在内。如果您想添加自己的头部并希望标准头部也映射，您还必须包含此标记或使用`header-mapper`提供您自己的`HeaderMapper`实现。
6	参见`<int-stomp:outbound-channel-adapter>`上的相同选项。
7	要订阅的 STOMP 目标名称的逗号分隔列表。目标列表（以及订阅）可以通过`addDestination()`和`removeDestination()@ManagedOperation`注释在运行时修改。
8	如果通道可以阻塞，则在将消息发送到通道时等待的最大时间（以毫秒为单位）。例如，如果`QueueChannel`已达到其最大容量，则它可以阻塞，直到有可用空间为止。
9	从传入的 STOMP 帧转换的目标`payload`的 Java 类型的完全限定名。它默认为`String.class`。
10	参见`<int-stomp:outbound-channel-adapter>`上的相同选项。
11	参见`<int-stomp:outbound-channel-adapter>`上的相同选项。