RSocket

RSocket 是一种用于字节流传输的二进制协议。 它通过单一连接上的异步消息传递，实现了对称的交互模型。

Spring Framework 的 spring-messaging 模块为 RSocket 请求方和响应方提供支持，既可用于客户端，也可用于服务端。 更多详情请参见 Spring Framework 参考文档中的 RSocket 部分，其中包含 RSocket 协议的概述。

RSocket 策略自动配置

Spring Boot 会自动配置 RSocketStrategies bean，提供编码和解码 RSocket 负载所需的全部基础设施。 默认情况下，自动配置会按如下顺序尝试配置：

  1. 使用 Jackson 的 CBOR 编解码器

  2. 使用 Jackson 的 JSON 编解码器

spring-boot-starter-rsocket 启动器同时提供了这两项依赖。 更多自定义方式请参见 Jackson 支持部分。

开发者可通过创建实现 RSocketStrategiesCustomizer 接口的 bean，来自定义 RSocketStrategies 组件。 注意其 @Order 很重要，因为它决定了编解码器的顺序。

RSocket 服务端自动配置

Spring Boot 提供了 RSocket 服务端自动配置。 所需依赖由 spring-boot-starter-rsocket 提供。

Spring Boot 支持通过 WebFlux 服务器以 WebSocket 方式暴露 RSocket，或启动独立的 RSocket 服务器。 这取决于应用类型及其配置。

对于 WebFlux 应用（即 WebApplicationType.REACTIVE 类型），仅当以下属性匹配时，RSocket 服务器才会集成到 Web 服务器中：

  • Properties

  • YAML

spring.rsocket.server.mapping-path=/rsocket
spring.rsocket.server.transport=websocket
spring:
  rsocket:
    server:
      mapping-path: "/rsocket"
      transport: "websocket"
仅支持通过 Reactor Netty 将 RSocket 集成到 Web 服务器，因为 RSocket 本身就是基于该库构建的。

或者，也可以作为独立的嵌入式服务器启动 RSocket TCP 或 WebSocket 服务器。 除依赖要求外，唯一必须的配置是为该服务器定义端口：

  • Properties

  • YAML

spring.rsocket.server.port=9898
spring:
  rsocket:
    server:
      port: 9898

Spring Messaging RSocket 支持

Spring Boot 会自动配置 RSocket 的 Spring Messaging 基础设施。

这意味着 Spring Boot 会创建 RSocketMessageHandler bean，用于处理应用的 RSocket 请求。

使用 RSocketRequester 调用 RSocket 服务

一旦服务端与客户端之间建立了 RSocket 通道，任一方都可以向对方发送或接收请求。

作为服务端，你可以在 RSocket @Controller 的任意处理方法中注入 RSocketRequester 实例。 作为客户端，则需先配置并建立 RSocket 连接。 Spring Boot 会自动配置 RSocketRequester.Builder，为其提供所需编解码器，并应用所有 RSocketConnectorConfigurer bean。

RSocketRequester.Builder 实例为原型 bean，意味着每个注入点都会获得一个新实例。 这是有意为之，因为该 builder 是有状态的，不应使用同一实例创建不同配置的 requester。

以下代码展示了典型示例：

  • Java

  • Kotlin

import reactor.core.publisher.Mono;

import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Service;

@Service
public class MyService {

	private final RSocketRequester rsocketRequester;

	public MyService(RSocketRequester.Builder rsocketRequesterBuilder) {
		this.rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898);
	}

	public Mono<User> someRSocketCall(String name) {
		return this.rsocketRequester.route("user").data(name).retrieveMono(User.class);
	}

}
import org.springframework.messaging.rsocket.RSocketRequester
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono

@Service
class MyService(rsocketRequesterBuilder: RSocketRequester.Builder) {

	private val rsocketRequester: RSocketRequester

	init {
		rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898)
	}

	fun someRSocketCall(name: String): Mono<User> {
		return rsocketRequester.route("user").data(name).retrieveMono(
			User::class.java
		)
	}

}