|
对于最新的稳定版本，请使用 Spring Integration 6.4.0！ |
STOMP 支持
Spring 集成版本 4.2 引入了 STOMP（简单文本定向消息传递协议）客户端支持。
它基于 Spring Framework 的消息模块、stomp 包的架构、基础设施和 API。
Spring 集成使用许多 Spring STOMP 组件（例如 和 ）。
有关更多信息，请参见 Spring Framework 参考手册中的 Spring Framework STOMP Support 一章。StompSessionStompClientSupport
您需要将此依赖项包含在您的项目中：
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stomp</artifactId>
<version>6.3.6</version>
</dependency>compile "org.springframework.integration:spring-integration-stomp:6.3.6"对于服务器端组件，您需要添加 a 和/或 dependencies。org.springframework:spring-websocketio.projectreactor.netty:reactor-netty
概述
要配置 STOMP，您应该从 STOMP 客户端对象开始。 Spring Framework 提供了以下实现：
-
WebSocketStompClient：基于 Spring WebSocket API 构建，支持标准 JSR-356 WebSocket、Jetty 9 和 Sockjs，用于使用 Sockjs 客户端进行基于 HTTP 的 WebSocket 仿真。 -
ReactorNettyTcpStompClient：基于项目构建。ReactorNettyTcpClientreactor-netty
您可以提供任何其他实现。
请参阅这些类的 Javadoc。StompClientSupport
该类被设计为一个工厂，用于为提供的 a 生成一个，所有剩余的工作都是通过对该 API 的回调和抽象完成的。
使用 Spring 集成适配器抽象，我们需要提供一些托管的共享对象，以将我们的应用程序表示为具有唯一会话的 STOMP 客户端。
为此， Spring Integration 提供了抽象来管理任何提供的 single 之间的 single 。
这允许对特定的 STOMP Broker 使用入站或出站通道适配器（或两者）。
有关更多信息，请参见（及其实现）JavaDocs。StompClientSupportStompSessionStompSessionHandlerStompSessionHandlerStompSessionStompSessionManagerStompSessionStompSessionHandlerStompSessionManager
STOMP 入站通道适配器
这是一个一站式组件，它将您的 Spring 集成应用程序订阅到提供的 STOMP 目标并从它们接收消息（通过使用 provided on the connected 从 STOMP 帧转换而来）。
您可以在运行时通过在 .StompInboundChannelAdapterMessageProducerMessageConverterStompSession@ManagedOperationStompInboundChannelAdapter
有关更多配置选项，请参阅 STOMP 名称空间支持和 Javadoc。StompInboundChannelAdapter
STOMP 出站通道适配器
这是 for the ，用于通过 （由共享 提供） 将传出实例发送到 STOMP （在运行时使用 SPEL 表达式预配置或确定）。StompMessageHandlerMessageHandler<int-stomp:outbound-channel-adapter>Message<?>destinationStompSessionStompSessionManager
有关更多配置选项，请参阅 STOMP 名称空间支持和 Javadoc。StompMessageHandler
STOMP 标头映射
STOMP 协议提供标头作为其帧的一部分。 STOMP 帧的整个结构具有以下格式：
....
COMMAND
header1:value1
header2:value2
Body^@
....Spring Framework 提供来表示这些 Headers。
有关更多详细信息，请参阅 Javadoc。
STOMP 帧与实例相互转换，这些 Headers 与实例相互映射。
Spring 集成为 STOMP 适配器提供了默认实现。
实现方式为 .
它分别为入站和出站适配器提供 和 操作。StompHeadersMessage<?>MessageHeadersHeaderMapperStompHeaderMapperfromHeaders()toHeaders()
与许多其他 Spring 集成模块一样，引入了该类以将标准 STOMP 头映射到，作为头名称前缀。
此外，具有该前缀的所有实例都会映射到 when sending to a destination.IntegrationStompHeadersMessageHeadersstomp_MessageHeadersStompHeaders
有关更多信息，请参阅这些类的 Javadoc 和 STOMP 名称空间支持中的属性描述。mapped-headers
STOMP 集成事件
许多 STOMP 操作是异步的，包括错误处理。
例如，STOMP 有一个服务器帧，当 Client 端帧通过添加 Headers 请求一个服务器帧时，它会返回该服务器帧。
为了提供对这些异步事件的访问， Spring 集成发出实例，你可以通过实现或使用（参见接收 Spring 应用程序事件）来获得这些实例。RECEIPTRECEIPTStompIntegrationEventApplicationListener<int-event:inbound-channel-adapter>
具体来说，当 a 由于无法连接到 STOMP 代理而接收时，会发出 a。
另一个示例是 .
它处理 STOMP 帧，这些帧是服务器对此发送的不当（未接受）消息的响应。StompExceptionEventAbstractStompSessionManagerstompSessionListenableFutureonFailure()StompMessageHandlerERRORStompMessageHandler
发出作为发送到 的消息的异步应答中回调的一部分。
可以是正数或负数，具体取决于该时间段内是否从服务器接收了帧，您可以在实例上配置该周期。
它默认为 （以毫秒为单位，因此为 15 秒）。StompMessageHandlerStompReceiptEventStompSession.ReceiptableStompSessionStompReceiptEventRECEIPTreceiptTimeLimitStompClientSupport15 * 1000
仅当要发送的消息的 STOMP 报头不是 时，才会添加回调。
您可以分别在 通过其选项和 上启用自动标头生成。StompSession.ReceiptableRECEIPTnullRECEIPTStompSessionautoReceiptStompSessionManager |
有关如何配置 Spring 集成以接受这些实例的更多信息，请参见STOMP 适配器 Java 配置。ApplicationEvent
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 集成 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 名称。
使用 bean 别名 plus 注册。
如果未设置该属性，则会在应用程序上下文中创建并注册 a，并将此属性的值作为 Bean 名称。
在这种情况下，端点是使用 bean name plus 注册的。MessageHandlerid.handlerchannelDirectChannelidid.adapter |
| 2 | 标识连接到此适配器的通道（如果存在）。
看。
自选。idid |
| 3 | 对 bean 的引用，它封装了低级连接和处理操作。
必填。StompSessionManagerStompSession |
| 4 | 对实现的 bean 的引用，该 bean 将 Spring Integration 映射到 Spring Integration 或从
STOMP 帧标头。
它与 互斥。
它默认为 .HeaderMapper<StompHeaders>MessageHeadersmapped-headersStompHeaderMapper |
| 5 | 要映射到 STOMP 帧标头的 STOMP 标头名称的逗号分隔列表。
仅当未设置引用时，才能提供它。
此列表中的值也可以是与 Headers 名称（如 或 ）匹配的简单模式。
特殊令牌 （） 表示所有标准 STOMP 标头（content-length、receipt、heart-beat 等）。
默认情况下，它们被包括在内。
如果要添加自己的标头并希望同时映射标准标头，则必须包含此令牌或使用 .header-mappermyheader**myheaderSTOMP_OUTBOUND_HEADERSHeaderMapperheader-mapper |
| 6 | STOMP 消息发送到的目标的名称。
它与 互斥。destination-expression |
| 7 | 一个 SPEL 表达式，要在运行时针对作为根对象的每个 Spring 集成进行评估。
它与 互斥。Messagedestination |
| 8 | 指示此端点是否应自动启动的布尔值。
它默认为 .true |
| 9 | 此终端节点应在其中启动和停止的生命周期阶段。
值越低，此终端节点开始得越早，停止得越晚。
默认值为 .
值可以是负数。
请参阅 SmartLifeCycle。Integer.MIN_VALUE |
了解元素<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 名称。
如果未设置该属性，则会在应用程序上下文中创建并注册 a，并将此属性的值作为 Bean 名称。
在这种情况下，端点使用 bean name plus 注册。channelDirectChannelidid.adapter |
| 2 | 标识连接到此适配器的通道。 |
| 3 | 应发送哪些实例的 bean 引用。MessageChannelErrorMessage |
| 4 | 在 <int-stomp：outbound-channel-adapter> 上查看相同的选项。 |
| 5 | 要从 STOMP 帧 Headers 映射的 STOMP Headers 名称的逗号分隔列表。
只有在未设置引用时，您才能提供此项。
此列表中的值也可以是与 Headers 名称匹配的简单模式（例如 或 ）。
特殊令牌 （） 表示所有标准 STOMP 标头（content-length、receipt、heart-beat 等）。
默认情况下，它们被包括在内。
如果要添加自己的标头并希望也映射标准标头，则还必须包含此令牌或使用 .header-mappermyheader**myheaderSTOMP_INBOUND_HEADERSHeaderMapperheader-mapper |
| 6 | 在 <int-stomp：outbound-channel-adapter> 上查看相同的选项。 |
| 7 | 要订阅的 STOMP 目标名称的逗号分隔列表。
目标列表（以及订阅）可以在运行时通过 和 注解进行修改。addDestination()removeDestination()@ManagedOperation |
| 8 | 如果通道可以阻塞，则向通道发送消息时要等待的最长时间（以毫秒为单位）。
例如，如果已达到最大容量，则 can 会阻塞，直到有可用空间。QueueChannel |
| 9 | 要从传入的 STOMP 帧转换的目标的 Java 类型的完全限定名称。
它默认为 .payloadString.class |
| 10 | 在 <int-stomp：outbound-channel-adapter> 上查看相同的选项。 |
| 11 | 在 <int-stomp：outbound-channel-adapter> 上查看相同的选项。 |