WebSocket

参考文档的这一部分涵盖了对响应式堆栈 WebSocket 消息传递的支持。

WebSocket 简介

WebSocket 协议，RFC 6455，提供了一种标准化方式，可以在单个 TCP 连接上建立客户端和服务器之间的全双工、双向通信通道。它与 HTTP 是不同的 TCP 协议，但设计为在 HTTP 上运行，使用端口 80 和 443，并允许重用现有的防火墙规则。

WebSocket 交互以使用 HTTP Upgrade 标头升级或在这种情况下切换到 WebSocket 协议的 HTTP 请求开始。以下示例显示了这种交互

GET /spring-websocket-portfolio/portfolio HTTP/1.1
Host: localhost:8080
Upgrade: websocket (1)
Connection: Upgrade (2)
Sec-WebSocket-Key: Uc9l9TMkWGbHFD2qnFHltg==
Sec-WebSocket-Protocol: v10.stomp, v11.stomp
Sec-WebSocket-Version: 13
Origin: https://127.0.0.1:8080
1 Upgrade 标头。
2 使用 Upgrade 连接。

支持 WebSocket 的服务器不会返回通常的 200 状态代码，而是返回类似于以下内容的输出

HTTP/1.1 101 Switching Protocols (1)
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: 1qVdfYHU9hPOl4JYYNXF623Gzn0=
Sec-WebSocket-Protocol: v10.stomp
1 协议切换

在成功握手后，HTTP 升级请求底层的 TCP 套接字将保持打开状态，以便客户端和服务器都可以继续发送和接收消息。

关于 WebSocket 工作原理的完整介绍超出了本文档的范围。请参阅 RFC 6455、HTML5 的 WebSocket 章节或网络上的许多介绍和教程。

请注意，如果 WebSocket 服务器在 Web 服务器（例如 nginx）后面运行，您可能需要对其进行配置以将 WebSocket 升级请求传递到 WebSocket 服务器。同样，如果应用程序在云环境中运行，请查看云提供商与 WebSocket 支持相关的说明。

HTTP 与 WebSocket

尽管 WebSocket 被设计为与 HTTP 兼容，并且以 HTTP 请求开始，但重要的是要理解这两种协议会导致截然不同的架构和应用程序编程模型。

在 HTTP 和 REST 中，应用程序被建模为多个 URL。为了与应用程序交互，客户端以请求-响应的方式访问这些 URL。服务器根据 HTTP URL、方法和标头将请求路由到相应的处理程序。

相比之下，在 WebSockets 中，通常只有一个 URL 用于初始连接。随后，所有应用程序消息都通过相同的 TCP 连接流动。这指向了一种完全不同的异步、事件驱动、消息传递架构。

WebSocket 也是一种低级传输协议，与 HTTP 不同，它没有为消息内容规定任何语义。这意味着，除非客户端和服务器就消息语义达成一致，否则无法路由或处理消息。

WebSocket 客户端和服务器可以通过 HTTP 握手请求中的 Sec-WebSocket-Protocol 标头协商使用更高级别的消息传递协议（例如 STOMP）。在没有该标头的情况下，它们需要制定自己的约定。

何时使用 WebSockets

WebSockets 可以使网页变得动态和交互式。但是，在许多情况下，AJAX 和 HTTP 流式传输或长轮询的组合可以提供一个简单有效的解决方案。

例如，新闻、邮件和社交信息流需要动态更新，但每隔几分钟更新一次可能完全可以。另一方面，协作、游戏和金融应用程序需要更接近实时。

延迟本身并不是决定因素。如果消息量相对较低（例如，监控网络故障），HTTP 流式传输或轮询可以提供有效的解决方案。正是低延迟、高频率和高容量的结合，才使得使用 WebSocket 成为最佳选择。

还要记住，在互联网上，不受您控制的限制性代理可能会阻止 WebSocket 交互，要么是因为它们没有配置为传递 Upgrade 标头，要么是因为它们关闭了看起来处于空闲状态的长期连接。这意味着，在防火墙内使用 WebSocket 进行内部应用程序的决策比面向公众的应用程序更直接。

WebSocket API

参见 Servlet 堆栈中的等效项

Spring 框架提供了一个 WebSocket API，您可以使用它来编写处理 WebSocket 消息的客户端和服务器端应用程序。

服务器

参见 Servlet 堆栈中的等效项

要创建 WebSocket 服务器，您可以先创建一个 WebSocketHandler。以下示例展示了如何做到这一点

  • Java

  • Kotlin

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;

public class MyWebSocketHandler implements WebSocketHandler {

	@Override
	public Mono<Void> handle(WebSocketSession session) {
		// ...
	}
}
import org.springframework.web.reactive.socket.WebSocketHandler
import org.springframework.web.reactive.socket.WebSocketSession

class MyWebSocketHandler : WebSocketHandler {

	override fun handle(session: WebSocketSession): Mono<Void> {
		// ...
	}
}

然后您可以将其映射到一个 URL

  • Java

  • Kotlin

@Configuration
class WebConfig {

	@Bean
	public HandlerMapping handlerMapping() {
		Map<String, WebSocketHandler> map = new HashMap<>();
		map.put("/path", new MyWebSocketHandler());
		int order = -1; // before annotated controllers

		return new SimpleUrlHandlerMapping(map, order);
	}
}
@Configuration
class WebConfig {

	@Bean
	fun handlerMapping(): HandlerMapping {
		val map = mapOf("/path" to MyWebSocketHandler())
		val order = -1 // before annotated controllers

		return SimpleUrlHandlerMapping(map, order)
	}
}

如果使用 WebFlux 配置，则无需进一步操作，否则，如果不使用 WebFlux 配置，则需要声明一个 WebSocketHandlerAdapter，如下所示

  • Java

  • Kotlin

@Configuration
class WebConfig {

	// ...

	@Bean
	public WebSocketHandlerAdapter handlerAdapter() {
		return new WebSocketHandlerAdapter();
	}
}
@Configuration
class WebConfig {

	// ...

	@Bean
	fun handlerAdapter() =  WebSocketHandlerAdapter()
}

WebSocketHandler

WebSocketHandler 的 handle 方法接受 WebSocketSession 并返回 Mono<Void> 以指示何时完成应用程序对会话的处理。会话通过两个流进行处理，一个用于入站消息，另一个用于出站消息。下表描述了处理流的两种方法

WebSocketSession 方法 描述

Flux<WebSocketMessage> receive()

提供对入站消息流的访问，并在连接关闭时完成。

Mono<Void> send(Publisher<WebSocketMessage>)

接受出站消息的源，写入消息，并返回一个 Mono<Void>，该 Mono<Void> 在源完成且写入完成时完成。

WebSocketHandler 必须将入站和出站流组合成一个统一的流，并返回一个 Mono<Void>，该 Mono<Void> 反映了该流的完成。根据应用程序要求，统一流在以下情况下完成

  • 入站或出站消息流完成。

  • 入站流完成（即连接关闭），而出站流是无限的。

  • 在通过 WebSocketSession 的 close 方法选择的点。

当入站和出站消息流组合在一起时，无需检查连接是否打开，因为 Reactive Streams 会发出结束活动信号。入站流接收完成或错误信号，出站流接收取消信号。

处理程序最基本的实现是处理入站流的实现。以下示例展示了这种实现

  • Java

  • Kotlin

class ExampleHandler implements WebSocketHandler {

	@Override
	public Mono<Void> handle(WebSocketSession session) {
		return session.receive()			(1)
				.doOnNext(message -> {
					// ...					(2)
				})
				.concatMap(message -> {
					// ...					(3)
				})
				.then();					(4)
	}
}
1 访问入站消息流。
2 对每条消息执行操作。
3 执行使用消息内容的嵌套异步操作。
4 返回一个Mono<Void>，当接收完成时完成。 
class ExampleHandler : WebSocketHandler {

	override fun handle(session: WebSocketSession): Mono<Void> {
		return session.receive()            (1)
				.doOnNext {
					// ...					(2)
				}
				.concatMap {
					// ...					(3)
				}
				.then()                     (4)
	}
}
1 访问入站消息流。
2 对每条消息执行操作。
3 执行使用消息内容的嵌套异步操作。
4 返回一个Mono<Void>，当接收完成时完成。
对于嵌套的异步操作，您可能需要在使用池化数据缓冲区的底层服务器（例如 Netty）上调用message.retain()。否则，数据缓冲区可能会在您有机会读取数据之前被释放。有关更多背景信息，请参阅数据缓冲区和编解码器。

以下实现结合了入站和出站流

  • Java

  • Kotlin

class ExampleHandler implements WebSocketHandler {

	@Override
	public Mono<Void> handle(WebSocketSession session) {

		Flux<WebSocketMessage> output = session.receive()				(1)
				.doOnNext(message -> {
					// ...
				})
				.concatMap(message -> {
					// ...
				})
				.map(value -> session.textMessage("Echo " + value));	(2)

		return session.send(output);									(3)
	}
}
1 处理入站消息流。
2 创建出站消息，生成一个组合流。
3 返回一个Mono<Void>，只要我们继续接收，它就不会完成。 
class ExampleHandler : WebSocketHandler {

	override fun handle(session: WebSocketSession): Mono<Void> {

		val output = session.receive()                      (1)
				.doOnNext {
					// ...
				}
				.concatMap {
					// ...
				}
				.map { session.textMessage("Echo $it") }    (2)

		return session.send(output)                         (3)
	}
}
1 处理入站消息流。
2 创建出站消息，生成一个组合流。
3 返回一个Mono<Void>，只要我们继续接收，它就不会完成。

入站和出站流可以是独立的，并且仅在完成时才连接，如下面的示例所示

  • Java

  • Kotlin

class ExampleHandler implements WebSocketHandler {

	@Override
	public Mono<Void> handle(WebSocketSession session) {

		Mono<Void> input = session.receive()								(1)
				.doOnNext(message -> {
					// ...
				})
				.concatMap(message -> {
					// ...
				})
				.then();

		Flux<String> source = ... ;
		Mono<Void> output = session.send(source.map(session::textMessage));	(2)

		return Mono.zip(input, output).then();								(3)
	}
}
1 处理入站消息流。
2 发送出站消息。
3 连接流并返回一个Mono<Void>，当任一流结束时完成。 
class ExampleHandler : WebSocketHandler {

	override fun handle(session: WebSocketSession): Mono<Void> {

		val input = session.receive()									(1)
				.doOnNext {
					// ...
				}
				.concatMap {
					// ...
				}
				.then()

		val source: Flux<String> = ...
		val output = session.send(source.map(session::textMessage))		(2)

		return Mono.zip(input, output).then()							(3)
	}
}
1 处理入站消息流。
2 发送出站消息。
3 连接流并返回一个Mono<Void>，当任一流结束时完成。

DataBuffer

DataBuffer是 WebFlux 中字节缓冲区的表示。参考手册的 Spring Core 部分在数据缓冲区和编解码器部分对此进行了更多介绍。需要理解的关键点是，在某些服务器（如 Netty）上，字节缓冲区是池化的并进行引用计数，必须在使用后释放它们以避免内存泄漏。

在 Netty 上运行时，应用程序必须使用DataBufferUtils.retain(dataBuffer)，如果它们希望保留输入数据缓冲区以确保它们不会被释放，并且随后在缓冲区被使用后使用DataBufferUtils.release(dataBuffer)。

握手

参见 Servlet 堆栈中的等效项

WebSocketHandlerAdapter委托给WebSocketService。默认情况下，这是一个HandshakeWebSocketService实例，它对 WebSocket 请求执行基本检查，然后使用正在使用的服务器的RequestUpgradeStrategy。目前，内置支持 Reactor Netty、Tomcat、Jetty 和 Undertow。

HandshakeWebSocketService公开了一个sessionAttributePredicate属性，它允许设置一个Predicate<String>来从WebSession中提取属性并将其插入到WebSocketSession的属性中。

服务器配置

参见 Servlet 堆栈中的等效项

每个服务器的RequestUpgradeStrategy都公开了特定于底层 WebSocket 服务器引擎的配置。当使用 WebFlux Java 配置时，您可以自定义这些属性，如WebFlux 配置的相应部分所示，或者，如果未使用 WebFlux 配置，请使用以下方法

  • Java

  • Kotlin

@Configuration
class WebConfig {

	@Bean
	public WebSocketHandlerAdapter handlerAdapter() {
		return new WebSocketHandlerAdapter(webSocketService());
	}

	@Bean
	public WebSocketService webSocketService() {
		TomcatRequestUpgradeStrategy strategy = new TomcatRequestUpgradeStrategy();
		strategy.setMaxSessionIdleTimeout(0L);
		return new HandshakeWebSocketService(strategy);
	}
}
@Configuration
class WebConfig {

	@Bean
	fun handlerAdapter() =
			WebSocketHandlerAdapter(webSocketService())

	@Bean
	fun webSocketService(): WebSocketService {
		val strategy = TomcatRequestUpgradeStrategy().apply {
			setMaxSessionIdleTimeout(0L)
		}
		return HandshakeWebSocketService(strategy)
	}
}

检查您服务器的升级策略，以查看有哪些选项可用。目前，只有 Tomcat 和 Jetty 公开了这些选项。

CORS

参见 Servlet 堆栈中的等效项

配置 CORS 并限制对 WebSocket 端点的访问的最简单方法是让您的WebSocketHandler实现CorsConfigurationSource并返回一个带有允许的来源、标头和其他详细信息的CorsConfiguration。如果您无法做到这一点，您也可以在SimpleUrlHandler上设置corsConfigurations属性，以通过 URL 模式指定 CORS 设置。如果两者都指定，则通过在CorsConfiguration上使用combine方法将它们组合起来。

客户端

Spring WebFlux 提供了一个WebSocketClient抽象，它为 Reactor Netty、Tomcat、Jetty、Undertow 和标准 Java（即 JSR-356）提供了实现。

Tomcat 客户端实际上是标准 Java 客户端的扩展，它在WebSocketSession处理中具有一些额外的功能，以利用 Tomcat 特定的 API 来暂停接收消息以进行背压。

要启动 WebSocket 会话，您可以创建客户端的实例并使用它的execute方法

  • Java

  • Kotlin

WebSocketClient client = new ReactorNettyWebSocketClient();

URI url = new URI("ws://127.0.0.1:8080/path");
client.execute(url, session ->
		session.receive()
				.doOnNext(System.out::println)
				.then());
val client = ReactorNettyWebSocketClient()

		val url = URI("ws://127.0.0.1:8080/path")
		client.execute(url) { session ->
			session.receive()
					.doOnNext(::println)
			.then()
		}

一些客户端，例如 Jetty，实现了Lifecycle，需要在使用它们之前停止和启动。所有客户端都具有与底层 WebSocket 客户端配置相关的构造函数选项。