STOMP 支持

STOMP 支持

Spring 集成版本 4.2 引入了 STOMP(简单文本定向消息传递协议)客户端支持。 它基于 Spring Framework 的消息模块、stomp 包的架构、基础设施和 API。 Spring 集成使用许多 Spring STOMP 组件(例如StompSessionStompClientSupport). 有关更多信息,请参见 Spring Framework 参考手册中的 Spring Framework STOMP Support 一章。spring-doc.cadn.net.cn

您需要将此依赖项包含在您的项目中:spring-doc.cadn.net.cn

Maven 系列
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stomp</artifactId>
    <version>6.0.9</version>
</dependency>
Gradle
compile "org.springframework.integration:spring-integration-stomp:6.0.9"

对于服务器端组件,您需要添加一个org.springframework:spring-websocket和/或io.projectreactor.netty:reactor-netty依赖。spring-doc.cadn.net.cn

概述

要配置 STOMP,您应该从 STOMP 客户端对象开始。 Spring Framework 提供了以下实现:spring-doc.cadn.net.cn

  • WebSocketStompClient:基于 Spring WebSocket API 构建,支持标准 JSR-356 WebSocket、Jetty 9 和 Sockjs,用于使用 Sockjs 客户端进行基于 HTTP 的 WebSocket 仿真。spring-doc.cadn.net.cn

  • ReactorNettyTcpStompClient:基于ReactorNettyTcpClientreactor-netty项目。spring-doc.cadn.net.cn

您可以提供任何其他StompClientSupport实现。 请参阅这些类的 Javadocspring-doc.cadn.net.cn

StompClientSupportclass 被设计为工厂来生成StompSession对于提供的StompSessionHandler其余所有工作都是通过回调完成的StompSessionHandlerStompSession抽象化。 使用 Spring 集成适配器抽象,我们需要提供一些托管的共享对象,以将我们的应用程序表示为具有唯一会话的 STOMP 客户端。 为此, Spring 集成提供了StompSessionManagerabstraction 来管理单个 StompSession之间提供的任何StompSessionHandler. 这允许对特定的 STOMP Broker 使用入站出站通道适配器(或两者)。 看StompSessionManager(及其实现)JavaDocs 了解更多信息。spring-doc.cadn.net.cn

STOMP 入站通道适配器

StompInboundChannelAdapter是一站式的MessageProducer组件将 Spring 集成应用程序订阅到提供的 STOMP 目标,并从中接收消息(使用提供的MessageConverter在互联StompSession). 您可以在运行时使用适当的@ManagedOperationannotations 上的StompInboundChannelAdapter.spring-doc.cadn.net.cn

有关更多配置选项,请参阅 STOMP 命名空间支持StompInboundChannelAdapter Javadoc 的 Java 文档。spring-doc.cadn.net.cn

STOMP 出站通道适配器

StompMessageHandlerMessageHandler对于<int-stomp:outbound-channel-adapter>,用于发送传出Message<?>实例添加到 STOMPdestination(预先配置或在运行时使用 SPEL 表达式确定)通过StompSession(由共享的StompSessionManager).spring-doc.cadn.net.cn

有关更多配置选项,请参阅 STOMP 命名空间支持StompMessageHandler Javadoc 的 Java 文档。spring-doc.cadn.net.cn

STOMP 标头映射

STOMP 协议提供标头作为其帧的一部分。 STOMP 帧的整个结构具有以下格式:spring-doc.cadn.net.cn

....
COMMAND
header1:value1
header2:value2

Body^@
....

Spring Framework 提供StompHeaders来表示这些标头。 有关更多详细信息,请参阅 Javadoc。 STOMP 帧相互转换Message<?>实例,这些标头映射到MessageHeaders实例。 Spring 集成提供了一个默认的HeaderMapperSTOMP 适配器的实现。 实现方式是StompHeaderMapper. 它提供fromHeaders()toHeaders()作。spring-doc.cadn.net.cn

与许多其他 Spring 集成模块一样,IntegrationStompHeaders类已引入以将标准 STOMP 标头映射到MessageHeadersstomp_作为标头名称前缀。 此外,所有MessageHeaders具有该前缀的实例将映射到StompHeaders发送到接收方时。spring-doc.cadn.net.cn

有关更多信息,请参阅这些类的 Javadocmapped-headers属性描述。spring-doc.cadn.net.cn

STOMP 集成事件

许多 STOMP作是异步的,包括错误处理。 例如,STOMP 有一个RECEIPTserver 帧,当客户端帧请求一个 server 帧时,它通过添加RECEIPT页眉。 为了提供对这些异步事件的访问, Spring 集成会发出StompIntegrationEvent实例,您可以通过实现ApplicationListener或使用<int-event:inbound-channel-adapter>(参见接收 Spring 应用程序事件)。spring-doc.cadn.net.cn

具体来说,StompExceptionEventAbstractStompSessionManagerstompSessionListenableFuture接收onFailure()由于无法连接到 STOMP 代理。 另一个示例是StompMessageHandler. 它处理ERRORSTOMP 帧,这是服务器对此发送的不当(未接受)消息的响应StompMessageHandler.spring-doc.cadn.net.cn

StompMessageHandler发出StompReceiptEvent作为StompSession.Receiptable异步应答中的回调,用于发送到StompSession. 这StompReceiptEvent可以是正数或负数,具体取决于RECEIPT帧是从receiptTimeLimit期间,您可以在StompClientSupport实例。 它默认为15 * 1000(以毫秒为单位,即 15 秒)。spring-doc.cadn.net.cn

StompSession.Receiptable仅当RECEIPT要发送的消息的 STOMP 标头不是null. 您可以启用自动RECEIPT标头生成StompSession通过其autoReceipt选项,并在StompSessionManager分别。

有关如何配置 Spring 集成以接受这些的更多信息,请参见STOMP Adapters Java ConfigurationApplicationEvent实例。spring-doc.cadn.net.cn

STOMP 适配器 Java 配置

以下示例显示了 STOMP 适配器的全面 Java 配置:spring-doc.cadn.net.cn

@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 集成 STOMP 名称空间实现了入站和出站通道适配器组件。 要将其包含在您的配置中,请在您的应用程序上下文配置文件中提供以下命名空间声明:spring-doc.cadn.net.cn

<?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 出站通道适配器的可用属性:spring-doc.cadn.net.cn

<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 StompSessionManagerbean,它封装了低级连接和StompSession处理作。 必填。
4 对实现HeaderMapper<StompHeaders>映射 Spring IntegrationMessageHeaders往返 STOMP 帧标头。 它与mapped-headers. 它默认为StompHeaderMapper.
5 要映射到 STOMP 帧标头的 STOMP 标头名称的逗号分隔列表。 仅当header-mapperreference 未设置。 此列表中的值也可以是与 Headers 名称匹配的简单模式(例如myheader**myheader). 特殊令牌 (STOMP_OUTBOUND_HEADERS) 表示所有标准 STOMP 标头(content-length、receipt、heart-beat 等)。 默认情况下,它们被包括在内。 如果要添加自己的标头并希望同时映射标准标头,则必须包含此令牌或提供自己的令牌HeaderMapper使用header-mapper.
6 STOMP 消息发送到的目标的名称。 它与destination-expression.
7 在运行时针对每个 Spring 集成进行评估的 SPEL 表达式Message作为根对象。 它与destination.
8 指示此端点是否应自动启动的布尔值。 它默认为true.
9 此终端节点应在其中启动和停止的生命周期阶段。 值越低,此终端节点开始得越早,停止得越晚。 默认值为Integer.MIN_VALUE. 值可以是负数。 看SmartLifeCycle.

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

下面的清单显示了 STOMP 入站通道适配器的可用属性:spring-doc.cadn.net.cn

<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 MessageChannelbean 引用ErrorMessage实例。
4 <int-stomp:outbound-channel-adapter>.
5 要从 STOMP 帧 Headers 映射的 STOMP Headers 名称的逗号分隔列表。 只有在header-mapperreference 未设置。 此列表中的值也可以是与 Headers 名称匹配的简单模式(例如myheader**myheader). 特殊令牌 (STOMP_INBOUND_HEADERS) 表示所有标准 STOMP 标头(content-length、receipt、heart-beat 等)。 默认情况下,它们被包括在内。 如果要添加自己的标头并希望也映射标准标头,则还必须包含此令牌或提供自己的令牌HeaderMapperimplementation usingheader-mapper.
6 <int-stomp:outbound-channel-adapter>.
7 要订阅的 STOMP 目标名称的逗号分隔列表。 目标列表(以及订阅)可以在运行时通过addDestination()removeDestination() @ManagedOperation附注。
8 如果通道可以阻塞,则向通道发送消息时要等待的最长时间(以毫秒为单位)。 例如,QueueChannel如果已达到其最大容量,则可以阻止,直到空间可用。
9 目标的 Java 类型的完全限定名称payload从传入的 STOMP 帧转换。 它默认为String.class.
10 <int-stomp:outbound-channel-adapter>.
11 <int-stomp:outbound-channel-adapter>.

APP信息