Spring Integration如何实现网络套接字功能支持？

本文共计3556个文字，预计阅读时间需要15分钟。

从4.1版本起，Spring Integration支持WebSocket。它基于Spring框架的模块，包括架构、基础设置和API。因此，Spring WebSocket的许多组件和配置选项可以在Spring Integration中重用。

从4.1版本开始，Spring Integration支持WebSocket。 它基于 Spring 框架模块中的架构、基础设施和 API。 因此，Spring WebSocket的许多组件（如或）和配置选项（如）可以在Spring Integration中重用。 有关更多信息，请参阅 Spring 框架参考手册中的Spring 框架 WebSocket 支持​章节。​​web-socket​​​​SubProtocolHandler​​​​WebSocketClient​​​​@EnableWebSocketMessageBroker​​

您需要将此依赖项包含在项目中：

<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-websocket</artifactId> <version>6.0.0</version></dependency>

对于服务器端，必须显式包含依赖项。​​org.springframework:spring-webmvc​​

Spring Framework WebSocket 基础架构基于 Spring 消息传递基础，并提供了一个基本的消息传递框架，该框架基于 Spring Integration 使用的相同实现和实现（以及一些 POJO 方法注释映射）。 因此，Spring Integration 可以直接参与 WebSocket 流程，即使没有 WebSocket 适配器也是如此。 为此，您可以使用适当的注释配置 Spring 集成，如以下示例所示：​​MessageChannel​​​​MessageHandler​​​​@MessagingGateway​​

@MessagingGateway@Controllerpublic interface WebSocketGateway { @MessageMapping("/greeting") @SendToUser("/queue/answer") @Gateway(requestChannel = "greetingChannel") String greeting(String payload);}

概述

由于 WebSocket 协议根据定义是流式传输的，并且我们可以同时向 WebSocket 发送和接收消息，因此我们可以处理适当的 ，无论是在客户端还是服务器端。 为了封装连接管理和注册表，提供了 和 实现。 由于WebSocket API及其在Spring Framework中的实现（具有许多扩展），服务器端和客户端都使用相同的类（当然，从Java的角度来看）。 因此，两端的大多数连接和注册表选项都是相同的。 这使我们能够重用许多配置项和基础结构挂钩，以便在服务器端和客户端构建 WebSocket 应用程序。 以下示例显示了组件如何同时满足这两种用途：​​WebSocketSession​​​​WebSocketSession​​​​IntegrationWebSocketContainer​​​​ClientWebSocketContainer​​​​ServerWebSocketContainer​​​​WebSocketSession​​

//Client side@Beanpublic WebSocketClient webSocketClient() { return new SockJsClient(Collections.singletonList(new WebSocketTransport(new JettyWebSocketClient())));}@Beanpublic IntegrationWebSocketContainer clientWebSocketContainer() { return new ClientWebSocketContainer(webSocketClient(), "ws://my.server.com/endpoint");}//Server side@Beanpublic IntegrationWebSocketContainer serverWebSocketContainer() { return new ServerWebSocketContainer("/endpoint").withSockJs();}

旨在实现双向消息传递，并且可以在入站和出站通道适配器之间共享（见下文），在使用单向（发送或接收）WebSocket 消息传递时，只能从其中一个适配器引用。 它可以在没有任何通道适配器的情况下使用，但在这种情况下，仅充当注册表。​​IntegrationWebSocketContainer​​​​IntegrationWebSocketContainer​​​​WebSocketSession​​

将内部注册为 . 它在目标供应商 WebSocket 容器中提供的和其他服务器 WebSocket 选项（例如 或 ）下执行此操作。 此配准是通过基础结构组件实现的，其作用与注释相同。 这意味着，通过使用（或应用程序上下文中的任何 Spring 集成命名空间），您可以省略声明，因为 Spring 集成基础结构会检测所有 WebSocket 端点。​​ServerWebSocketContainer​​​​WebSocketConfigurer​​​​IntegrationWebSocketContainer.IntegrationWebSocketHandler​​​​Endpoint​​​​paths​​​​HandshakeHandler​​​​SockJS fallback​​​​ServletWebSocketHandlerRegistry​​​​WebSocketIntegrationConfigurationInitializer​​​​@EnableWebSocket​​​​@EnableIntegration​​​​@EnableWebSocket​​

WebSocket 入站通道适配器

实现交互的接收部分。 必须为其提供 ，适配器将自身注册为 以处理传入的消息和事件。​​WebSocketInboundChannelAdapter​​​​WebSocketSession​​​​IntegrationWebSocketContainer​​​​WebSocketListener​​​​WebSocketSession​​

只能在 中注册一个。​​WebSocketListener​​​​IntegrationWebSocketContainer​​

对于 WebSocket 子协议，可以将 配置为第二个构造函数参数。 适配器委托给 以确定适合接受的并根据子协议实现将 a 转换为 a。​​WebSocketInboundChannelAdapter​​​​SubProtocolHandlerRegistry​​​​SubProtocolHandlerRegistry​​​​SubProtocolHandler​​​​WebSocketSession​​​​WebSocketMessage​​​​Message​​

默认情况下，仅依赖于原始实现，这会将 .​​WebSocketInboundChannelAdapter​​​​PassThruSubProtocolHandler​​​​WebSocketMessage​​​​Message​​

仅接受具有空标头或空标头的实例并将其发送到底层集成流。 所有其他类型都通过实现发出的实例（例如 ）进行处理。​​WebSocketInboundChannelAdapter​​​​Message​​​​SimpMessageType.MESSAGE​​​​simpMessageType​​​​Message​​​​ApplicationEvent​​​​SubProtocolHandler​​​​StompSubProtocolHandler​​

在服务器端，如果存在配置，则可以使用该选项进行配置。 在这种情况下，所有类型都委托给提供的 . 此外，如果代理中继配置了目标前缀，那么与代理目标匹配的那些消息将路由到 的 ，而不是 的 。​​@EnableWebSocketMessageBroker​​​​WebSocketInboundChannelAdapter​​​​useBroker = true​​​​non-MESSAGE​​​​Message​​​​AbstractBrokerMessageHandler​​​​AbstractBrokerMessageHandler​​​​outputChannel​​​​WebSocketInboundChannelAdapter​​

如果接收到的消息属于该类型，则立即向 发送消息而不将其发送到通道。​​useBroker = false​​​​SimpMessageType.CONNECT​​​​WebSocketInboundChannelAdapter​​​​SimpMessageType.CONNECT_ACK​​​​WebSocketSession​​

Spring 的 WebSocket 支持只允许配置一个代理中继。 因此，我们不需要参考。 它在应用程序上下文中检测到。​​AbstractBrokerMessageHandler​​

有关更多配置选项，请参阅WebSockets 命名空间支持。

WebSocket 出站通道适配器

这：​​WebSocketOutboundChannelAdapter​​

接受来自其 的 Spring 集成消息MessageChannel

确定从WebSocketSessionidMessageHeaders

从提供的WebSocketSessionIntegrationWebSocketContainer

将工作的转换和发送委托给提供的 适当的 。WebSocketMessageSubProtocolHandlerSubProtocolHandlerRegistry

在客户端，消息标头不是必需的，因为仅处理单个连接及其分别。​​WebSocketSession​​​​id​​​​ClientWebSocketContainer​​​​WebSocketSession​​

要使用 STOMP 子协议，应将此适配器配置为 . 然后，您可以使用 和 将任何 STOMP 消息类型发送到此适配器，或者仅使用 （请参阅标头扩充器）。​​StompSubProtocolHandler​​​​StompHeaderAccessor.create(StompCommand…)​​​​MessageBuilder​​​​HeaderEnricher​​

本章的其余部分将介绍大部分其他配置选项。

WebSockets 命名空间支持

Spring Integration WebSocket 命名空间包括本章其余部分描述的几个组件。 若要将其包含在配置中，请在应用程序上下文配置文件中使用以下命名空间声明：

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

​​<int-websocket:client-container>​​属性

以下清单显示了可用于该元素的属性：​​<int-websocket:client-container>​​

<int-websocket:client-container id="" client="" uri="" uri-variables="" origin="" send-time-limit="" send-buffer-size-limit="" auto-startup="" phase=""> <int-websocket:d1fxtkz8shb9d2.cloudfront.net/sockjs-0.3.4.min.js​​​​/sockjs​​​​/sockjs/iframe.html​​

在关闭单个 HTTP 流请求之前，可以通过该请求发送的最小字节数。 默认值为 （即 128*1024 或 131072 字节）。​​128K​​

来自 SockJs 终结点的响应中的值。 此属性指示应用程序正常运行是否需要 Cookie（例如，用于负载平衡或在 Java Servlet 容器中使用 HTTP 会话）。​​cookie_needed​​​​/info​​​​JSESSIONID​​

服务器未发送任何消息的时间量（以毫秒为单位），之后服务器应 向客户端发送检测信号帧，以防止连接中断。 默认值为 （25 秒）。​​25,000​​

客户端在没有接收连接（即服务器可以通过其向客户端发送数据的活动连接）后被视为断开连接的时间量（以毫秒为单位）。 默认值为 。​​5000​​

会话在等待来自客户端的下一个 HTTP 轮询请求时可以缓存的服务器到客户端消息数。 默认大小为 。​​100​​

某些负载均衡器不支持 WebSocket。 将此选项设置为 以禁用服务器端的 WebSocket 传输。 默认值为 。​​false​​​​true​​

豆引用。 如果未提供任何值，则会创建一个新实例。 此计划程序实例用于计划检测信号消息。​​TaskScheduler​​​​ThreadPoolTaskScheduler​​

Thebean reference to use for encoding and decoding SockJS messages. By default,is used, which requires the Jackson library to be present on the classpath.​​SockJsMessageCodec​​​​Jackson2SockJsMessageCodec​​

List ofbean references.​​TransportHandler​​

Whether to disable automatic addition of CORS headers for SockJS requests. The default value is.​​false​​

​​<int-websocket:outbound-channel-adapter>​​属性

以下清单显示了可用于该元素的属性：​​<int-websocket:outbound-channel-adapter>​​

<int-websocket:outbound-channel-adapter id="" channel="" container="" default-protocol-handler="" protocol-handlers="" message-converters="" merge-with-default-converters="" auto-startup="" phase=""/>

组件 Bean 名称。 如果未提供该属性，则会在应用程序上下文中创建并注册 ，并将此属性作为 Bean 名称。 在这种情况下，端点使用 Bean 名称加上 . 并且注册了 bean 别名加上 .​​channel​​​​DirectChannel​​​​id​​​​id​​​​.adapter​​​​MessageHandler​​​​id​​​​.handler​​

标识连接到此适配器的通道。

对 Bean 的引用，它封装了低级别的连接和处理操作。 必填。​​IntegrationWebSocketContainer​​​​WebSocketSession​​

对实例的可选引用。 当客户端未请求子协议或它是单个协议处理程序时，使用它。 如果未提供此引用或列表，则默认使用 。​​SubProtocolHandler​​​​protocol-handlers​​​​PassThruSubProtocolHandler​​

此通道适配器的 Bean 引用列表。 如果只提供单个 Bean 引用而不提供 ，则该单个引用将用作 。 如果未设置此属性或 ，则默认使用 。​​SubProtocolHandler​​​​default-protocol-handler​​​​SubProtocolHandler​​​​default-protocol-handler​​​​default-protocol-handler​​​​PassThruSubProtocolHandler​​

此通道适配器的 Bean 引用列表。​​MessageConverter​​

指示是否应在任何自定义转换器之后注册默认转换器的布尔值。 仅当提供了 时，才使用此标志。 否则，将注册所有默认转换器。 默认值为 。 缺省转换器是（按顺序）：、 和（如果类路径上存在杰克逊库）。​​message-converters​​​​false​​​​StringMessageConverter​​​​ByteArrayMessageConverter​​​​MappingJackson2MessageConverter​​

指示此终结点是否应自动启动的布尔值。 默认值为 。​​true​​

此终结点应在其中启动和停止的生命周期阶段。 值越低，此终结点启动得越早，停止得越晚。 缺省值为 。 值可以为负数。 请参阅智能生命周期​。​​Integer.MIN_VALUE​​

​​<int-websocket:inbound-channel-adapter>​​属性

以下清单显示了可用于该元素的属性：​​<int-websocket:outbound-channel-adapter>​​

<int-websocket:inbound-channel-adapter id="" channel="" error-channel="" container="" default-protocol-handler="" protocol-handlers="" message-converters="" merge-with-default-converters="" send-timeout="" payload-type="" use-broker="" auto-startup="" phase=""/>

组件 Bean 名称。 如果未设置该属性，则会在应用程序上下文中创建并注册 ，并将此属性作为 Bean 名称。 在这种情况下，端点使用 Bean 名称加上 .​​channel​​​​DirectChannel​​​​id​​​​id​​​​.adapter​​

标识连接到此适配器的通道。

实例应发送到的 Bean 引用。​​MessageChannel​​​​ErrorMessage​​

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

如果通道可以阻止，则向通道发送消息时等待的最长时间（以毫秒为单位）。 例如，如果达到其最大容量，则 can 会阻塞，直到空间可用。​​QueueChannel​​

要从传入 转换的目标的 Java 类型的完全限定名称。 默认值为 。​​payload​​​​WebSocketMessage​​​​java.lang.String​​

指示此适配器是否将具有代理目标的实例和消息从应用程序上下文发送到 。 当此属性为 时，需要配置。 此属性仅在服务器端使用。 在客户端，它被忽略。 默认值为 。​​non-MESSAGE​​​​WebSocketMessage​​​​AbstractBrokerMessageHandler​​​​true​​​​Broker Relay​​​​false​​

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

在<int-websocket：outbound-channel-adapter>上查看相同的选项。

用​​ClientStompEncoder​​

从版本 4.3.13 开始，Spring 集成提供了（作为标准的扩展）用于 WebSocket 通道适配器的客户端。 为了正确准备客户端消息，必须将 的实例注入到 . 默认值的一个问题是它是为服务器端设计的，因此它将标头更新为（根据服务器端的 STOMP 协议的要求）。 如果客户端未在正确的 Web 套接字帧中发送其消息，则某些 STOMP 代理不接受它们。 在本例中，的用途是在将消息编码到 .​​ClientStompEncoder​​​​StompEncoder​​​​ClientStompEncoder​​​​StompSubProtocolHandler​​​​StompSubProtocolHandler​​​​SEND​​​​stompCommand​​​​MESSAGE​​​​SEND​​​​ClientStompEncoder​​​​stompCommand​​​​SEND​​​​byte[]​​

动态 WebSocket 端点注册

从版本 5.5 开始，现在可以在运行时注册（和删除）WebSocket 服务器终结点（基于 ） 的通道适配器 - 映射的 a 通过 公开到 Web Socket 客户端可以访问。动态和运行时集成流支持有助于以透明的方式注册这些端点：​​ServerWebSocketContainer​​​​paths​​​​ServerWebSocketContainer​​​​HandlerMapping​​​​DispatcherServlet​​

@AutowiredIntegrationFlowContext integrationFlowContext;@AutowiredHandshakeHandler handshakeHandler;...ServerWebSocketContainer serverWebSocketContainer = new ServerWebSocketContainer("/dynamic") .setHandshakeHandler(this.handshakeHandler);WebSocketInboundChannelAdapter webSocketInboundChannelAdapter = new WebSocketInboundChannelAdapter(serverWebSocketContainer);QueueChannel dynamicRequestsChannel = new QueueChannel();IntegrationFlow serverFlow = IntegrationFlow.from(webSocketInboundChannelAdapter) .channel(dynamicRequestsChannel) .get();IntegrationFlowContext.IntegrationFlowRegistration dynamicServerFlow = this.integrationFlowContext.registration(serverFlow) .addBean(serverWebSocketContainer) .register();...dynamicServerFlow.destroy();

调用动态流注册以将 的实例添加到 for 终结点注册中非常重要。 销毁动态流注册时，也会销毁关联的实例以及相应的终结点注册，包括 URL 路径映射。​​.addBean(serverWebSocketContainer)​​​​ServerWebSocketContainer​​​​ApplicationContext​​​​ServerWebSocketContainer​​

动态 Websocket 端点只能通过 Spring 集成机制注册：当使用常规 Spring 时，Spring 集成配置会回退，并且不会注册动态端点的基础设施。​​@EnableWebsocket​​