RSocket
本节介绍 Spring Framework 对 RSocket 协议的支持。
概述
RSocket 是一种应用程序协议,用于通过 TCP、WebSocket 和其他字节流传输进行多路复用、双工通信,使用以下交互模型之一:
-
Request-Response
— 发送一条消息并接收一条。 -
Request-Stream
— 发送一条消息并接收一个消息流。 -
Channel
— 在两个方向发送消息流。 -
Fire-and-Forget
— 发送一条单向消息。
一旦建立初始连接,"客户端"与"服务器"的区别就会消失,因为双方变得对称,每一方都可以启动上述交互之一。这就是为什么在协议调用中,参与方被称为"请求者"和"响应者",而上述交互被称为"请求流"或简称为"请求"。
以下是 RSocket 协议的主要特性和优势:
-
Reactive Streams 跨网络边界的语义 — 对于
Request-Stream
和Channel
等流式请求,背压信号在请求者和响应者之间传递,允许请求者在源头减慢响应者,从而减少对网络层拥塞控制的依赖,以及在网络层或任何层进行缓冲的需求。 -
请求节流 — 此功能命名为"租约",因为它可以通过
LEASE
帧从每一端发送,以限制另一端在给定时间内允许的总请求数。租约会定期续订。 -
会话恢复 — 这旨在解决连接丢失问题,需要维护一些状态。状态管理对应用程序是透明的,并且与背压结合使用效果良好,背压可以在可能的情况下停止生产者并减少所需的状态量。
-
大消息的碎片化和重组。
-
心跳(keepalive)。
RSocket 在多种语言中都有 实现。 Java 库 基于 Project Reactor 构建,并使用 Reactor Netty 进行传输。这意味着您的应用程序中 Reactive Streams Publishers 的信号会通过 RSocket 透明地跨网络传播。
协议
RSocket 的优点之一是它具有定义明确的线上传输行为和易于阅读的 规范 以及一些协议 扩展。因此,阅读规范是一个好主意,而无需考虑语言实现和更高级别的框架 API。本节提供了一个简洁的概述,以建立一些上下文。
连接
最初,客户端通过 TCP 或 WebSocket 等低级流传输连接到服务器,并向服务器发送 SETUP
帧以设置连接参数。
服务器可能会拒绝 SETUP
帧,但通常在发送(对于客户端)和接收(对于服务器)之后,双方都可以开始发出请求,除非 SETUP
指示使用租约语义来限制请求数量,在这种情况下,双方必须等待来自另一端的 LEASE
帧以允许发出请求。
发出请求
一旦建立连接,双方都可以通过 REQUEST_RESPONSE
、REQUEST_STREAM
、REQUEST_CHANNEL
或 REQUEST_FNF
帧发起请求。每个帧都携带一条从请求者到响应者的消息。
然后,响应者可以返回带有响应消息的 PAYLOAD
帧,在 REQUEST_CHANNEL
的情况下,请求者也可以发送带有更多请求消息的 PAYLOAD
帧。
当请求涉及消息流(例如 Request-Stream
和 Channel
)时,响应者必须遵守来自请求者的需求信号。需求表示为消息数量。初始需求在 REQUEST_STREAM
和 REQUEST_CHANNEL
帧中指定。后续需求通过 REQUEST_N
帧发出信号。
双方还可以通过 METADATA_PUSH
帧发送元数据通知,这些通知不属于任何单个请求,而是针对整个连接。
消息格式
RSocket 消息包含数据和元数据。元数据可用于发送路由、安全令牌等。数据和元数据可以以不同的格式进行格式化。每种数据的 mime 类型在 SETUP
帧中声明,并适用于给定连接上的所有请求。
虽然所有消息都可以包含元数据,但通常路由等元数据是按请求的,因此仅包含在请求的第一条消息中,即与 REQUEST_RESPONSE
、REQUEST_STREAM
、REQUEST_CHANNEL
或 REQUEST_FNF
帧之一一起。
协议扩展定义了用于应用程序的常见元数据格式:
Java 实现
RSocket 的 Java 实现 基于 Project Reactor 构建。TCP 和 WebSocket 的传输基于 Reactor Netty 构建。作为一个 Reactive Streams 库,Reactor 简化了协议的实现工作。对于应用程序来说,使用 Flux
和 Mono
配合声明式操作符和透明的背压支持是自然的选择。
RSocket Java 中的 API 故意保持最小和基本。它专注于协议特性,并将应用程序编程模型(例如,RPC 代码生成与其他)作为一个更高级别、独立的问题。
主要契约 io.rsocket.RSocket 使用 Mono
表示单个消息的承诺,Flux
表示消息流,io.rsocket.Payload
表示实际消息,可访问数据和元数据作为字节缓冲区,对四种请求交互类型进行建模。RSocket
契约是对称使用的。对于请求,应用程序获得一个 RSocket
来执行请求。对于响应,应用程序实现 RSocket
来处理请求。
这并非详尽的介绍。在大多数情况下,Spring 应用程序无需直接使用其 API。但是,独立于 Spring 查看或试验 RSocket 可能很重要。RSocket Java 存储库包含许多 示例应用程序,演示其 API 和协议特性。
Spring 支持
spring-messaging
模块包含以下内容:
-
rsocket-requester — 用于通过
io.rsocket.RSocket
发出请求的流畅 API,带有数据和元数据编码/解码。 -
rsocket-annot-responders — 用于响应的
@MessageMapping
和@RSocketExchange
注解处理程序方法。 -
rsocket-interface — 作为 Java 接口的 RSocket 服务声明,带有
@RSocketExchange
方法,用作请求者或响应者。
spring-web
模块包含 Encoder
和 Decoder
实现,例如 Jackson CBOR/JSON 和 Protobuf,RSocket 应用程序可能需要这些实现。它还包含 PathPatternParser
,可以插入以实现高效的路由匹配。
Spring Boot 2.2 支持通过 TCP 或 WebSocket 启动 RSocket 服务器,包括在 WebFlux 服务器中通过 WebSocket 公开 RSocket 的选项。还提供了客户端支持和 RSocketRequester.Builder
和 RSocketStrategies
的自动配置。有关更多详细信息,请参阅 Spring Boot 参考手册中的
RSocket 部分。
Spring Security 5.2 提供 RSocket 支持。
Spring Integration 5.2 提供入站和出站网关,用于与 RSocket 客户端和服务器交互。有关更多详细信息,请参阅 Spring Integration 参考手册。
Spring Cloud Gateway 支持 RSocket 连接。
RSocketRequester
RSocketRequester
提供了一个流畅的 API 来执行 RSocket 请求,接受并返回用于数据和元数据的对象,而不是低级数据缓冲区。它可以对称使用,用于从客户端发出请求和从服务器发出请求。
客户端请求者
在客户端获取 RSocketRequester
是连接到服务器,这涉及到发送一个带有连接设置的 RSocket SETUP
帧。RSocketRequester
提供了一个构建器,有助于准备 io.rsocket.core.RSocketConnector
,包括 SETUP
帧的连接设置。
这是使用默认设置连接的最基本方式:
-
Java
-
Kotlin
RSocketRequester requester = RSocketRequester.builder().tcp("localhost", 7000);
URI url = URI.create("https://example.org:8080/rsocket");
RSocketRequester requester = RSocketRequester.builder().webSocket(url);
val requester = RSocketRequester.builder().tcp("localhost", 7000)
URI url = URI.create("https://example.org:8080/rsocket");
val requester = RSocketRequester.builder().webSocket(url)
上述操作不会立即连接。当发出请求时,将透明地建立并使用一个共享连接。
连接设置
RSocketRequester.Builder
提供以下内容来自定义初始 SETUP
帧:
-
dataMimeType(MimeType)
— 设置连接上数据的 mime 类型。 -
metadataMimeType(MimeType)
— 设置连接上元数据的 mime 类型。 -
setupData(Object)
— 包含在SETUP
中的数据。 -
setupRoute(String, Object…)
— 包含在SETUP
中的元数据中的路由。 -
setupMetadata(Object, MimeType)
— 包含在SETUP
中的其他元数据。
对于数据,默认的 mime 类型是从第一个配置的 Decoder
派生的。对于元数据,默认的 mime 类型是 复合元数据,它允许每个请求有多个元数据值和 mime 类型对。通常两者都不需要更改。
SETUP
帧中的数据和元数据是可选的。在服务器端,<<@ConnectMapping,rsocket-annot-connectmapping>> 方法可用于处理连接的开始和 SETUP
帧的内容。元数据可用于连接级别的安全性。
策略
RSocketRequester.Builder
接受 RSocketStrategies
来配置请求者。您需要使用它来提供编码器和解码器,用于数据和元数据值的(反)序列化。默认情况下,只注册了 spring-core
中用于 String
、byte[]
和 ByteBuffer
的基本编解码器。添加 spring-web
可以访问更多可以按如下方式注册的编解码器:
-
Java
-
Kotlin
RSocketStrategies strategies = RSocketStrategies.builder()
.encoders(encoders -> encoders.add(new Jackson2CborEncoder()))
.decoders(decoders -> decoders.add(new Jackson2CborDecoder()))
.build();
RSocketRequester requester = RSocketRequester.builder()
.rsocketStrategies(strategies)
.tcp("localhost", 7000);
val strategies = RSocketStrategies.builder()
.encoders { it.add(Jackson2CborEncoder()) }
.decoders { it.add(Jackson2CborDecoder()) }
.build()
val requester = RSocketRequester.builder()
.rsocketStrategies(strategies)
.tcp("localhost", 7000)
RSocketStrategies
旨在重复使用。在某些情况下,例如,客户端和服务器在同一个应用程序中,可能更倾向于在 Spring 配置中声明它。
客户端响应者
RSocketRequester.Builder
可用于配置响应服务器请求的响应者。
您可以将基于与服务器端使用的相同基础设施的注解处理程序用于客户端响应,但需要按如下方式以编程方式注册:
-
Java
-
Kotlin
RSocketStrategies strategies = RSocketStrategies.builder()
.routeMatcher(new PathPatternRouteMatcher()) [id="CO1-1"][id="CO1-1"](1)
.build();
SocketAcceptor responder =
RSocketMessageHandler.responder(strategies, new ClientHandler()); [id="CO1-2"][id="CO1-2"](2)
RSocketRequester requester = RSocketRequester.builder()
.rsocketConnector(connector -> connector.acceptor(responder)) [id="CO1-3"][id="CO1-3"](3)
.tcp("localhost", 7000);
1 | 如果存在 spring-web ,则使用 PathPatternRouteMatcher 进行高效路由匹配。 |
2 | 从带有 @MessageMapping 和/或 @ConnectMapping 方法的类创建响应者。 |
3 | 注册响应者。 |
val strategies = RSocketStrategies.builder()
.routeMatcher(PathPatternRouteMatcher()) [id="CO2-1"][id="CO1-4"](1)
.build()
val responder =
RSocketMessageHandler.responder(strategies, ClientHandler()); [id="CO2-2"][id="CO1-5"](2)
val requester = RSocketRequester.builder()
.rsocketConnector { it.acceptor(responder) } [id="CO2-3"][id="CO1-6"](3)
.tcp("localhost", 7000)
1 | 如果存在 spring-web ,则使用 PathPatternRouteMatcher 进行高效路由匹配。 |
2 | 从带有 @MessageMapping 和/或 @ConnectMapping 方法的类创建响应者。 |
3 | 注册响应者。 |
请注意,上述方法仅是为客户端响应者的编程注册设计的快捷方式。对于替代场景,即客户端响应者位于 Spring 配置中,您仍然可以将 RSocketMessageHandler
声明为 Spring bean,然后按如下方式应用:
-
Java
-
Kotlin
ApplicationContext context = ... ;
RSocketMessageHandler handler = context.getBean(RSocketMessageHandler.class);
RSocketRequester requester = RSocketRequester.builder()
.rsocketConnector(connector -> connector.acceptor(handler.responder()))
.tcp("localhost", 7000);
import org.springframework.beans.factory.getBean
val context: ApplicationContext = ...
val handler = context.getBean<RSocketMessageHandler>()
val requester = RSocketRequester.builder()
.rsocketConnector { it.acceptor(handler.responder()) }
.tcp("localhost", 7000)
对于上述情况,您可能还需要在 RSocketMessageHandler
中使用 setHandlerPredicate
来切换到不同的策略来检测客户端响应者,例如,基于自定义注解(如 @RSocketClientResponder
)而不是默认的 @Controller
。这在客户端和服务器或同一应用程序中存在多个客户端的场景中是必需的。
另请参阅 rsocket-annot-responders,了解有关编程模型的更多信息。
高级
RSocketRequesterBuilder
提供了一个回调来公开底层的 io.rsocket.core.RSocketConnector
,以获取更多配置选项,例如 keepalive 间隔、会话恢复、拦截器等。您可以在该级别配置选项,如下所示:
-
Java
-
Kotlin
RSocketRequester requester = RSocketRequester.builder()
.rsocketConnector(connector -> {
// ...
})
.tcp("localhost", 7000);
val requester = RSocketRequester.builder()
.rsocketConnector {
//...
}
.tcp("localhost", 7000)
服务器请求者
要从服务器向连接的客户端发出请求,只需从服务器获取连接客户端的请求者即可。
在 rsocket-annot-responders 中,@ConnectMapping
和 @MessageMapping
方法支持 RSocketRequester
参数。使用它来访问连接的请求者。请记住,@ConnectMapping
方法本质上是 SETUP
帧的处理程序,必须在请求开始之前处理。因此,最开始的请求必须与处理解耦。例如:
- Java
-
@ConnectMapping Mono<Void> handle(RSocketRequester requester) { requester.route("status").data("5") .retrieveFlux(StatusReport.class) .subscribe(bar -> { [id="CO3-1"][id="CO1-7"](1) // ... }); return ... ; [id="CO3-2"][id="CO1-8"](2) }
<1> 异步启动请求,独立于处理。 <1> 执行处理并返回完成 `Mono<Void>`。
- Kotlin
-
@ConnectMapping suspend fun handle(requester: RSocketRequester) { GlobalScope.launch { requester.route("status").data("5").retrieveFlow<StatusReport>().collect { [id="CO4-1"][id="CO2-1"](1) // ... } } /// ... [id="CO4-2"][id="CO2-2"](2) }
<1> 异步启动请求,独立于处理。 <1> 在挂起函数中执行处理。
请求
一旦您拥有 rsocket-requester-client 或 rsocket-requester-server 请求者,您可以按如下方式发出请求:
- Java
-
ViewBox viewBox = ... ; Flux<AirportLocation> locations = requester.route("locate.radars.within") [id="CO5-1"][id="CO3-1"](1) .data(viewBox) [id="CO5-2"][id="CO3-2"](2) .retrieveFlux(AirportLocation.class); [id="CO5-3"][id="CO3-3"](3)
<1> 指定要包含在请求消息元数据中的路由。 <1> 为请求消息提供数据。 <1> 声明预期的响应。
- Kotlin
-
val viewBox: ViewBox = ... val locations = requester.route("locate.radars.within") [id="CO6-1"][id="CO4-1"](1) .data(viewBox) [id="CO6-2"][id="CO4-2"](2) .retrieveFlow<AirportLocation>() [id="CO6-3"][id="CO4-3"](3)
<1> 指定要包含在请求消息元数据中的路由。 <1> 为请求消息提供数据。 <1> 声明预期的响应。
交互类型由输入和输出的基数隐式决定。上面的示例是 Request-Stream
,因为发送了一个值并接收了一个值流。在大多数情况下,只要输入和输出的选择与 RSocket 交互类型以及响应者预期的输入和输出类型匹配,您就不需要考虑这一点。唯一无效组合的示例是多对一。
data(Object)
方法还接受任何 Reactive Streams Publisher
,包括 Flux
和 Mono
,以及在 ReactiveAdapterRegistry
中注册的任何其他值生产者。对于产生相同类型值的多值 Publisher
(例如 Flux
),请考虑使用重载的 data
方法之一,以避免在每个元素上进行类型检查和 Encoder
查找:
data(Object producer, Class<?> elementClass);
data(Object producer, ParameterizedTypeReference<?> elementTypeRef);
data(Object)
步骤是可选的。对于不发送数据的请求,跳过此步骤:
-
Java
-
Kotlin
Mono<AirportLocation> location = requester.route("find.radar.EWR"))
.retrieveMono(AirportLocation.class);
import org.springframework.messaging.rsocket.retrieveAndAwait
val location = requester.route("find.radar.EWR")
.retrieveAndAwait<AirportLocation>()
如果使用 复合元数据(默认)并且值受注册的 Encoder
支持,则可以添加额外的元数据值。例如:
-
Java
-
Kotlin
String securityToken = ... ;
ViewBox viewBox = ... ;
MimeType mimeType = MimeType.valueOf("message/x.rsocket.authentication.bearer.v0");
Flux<AirportLocation> locations = requester.route("locate.radars.within")
.metadata(securityToken, mimeType)
.data(viewBox)
.retrieveFlux(AirportLocation.class);
import org.springframework.messaging.rsocket.retrieveFlow
val requester: RSocketRequester = ...
val securityToken: String = ...
val viewBox: ViewBox = ...
val mimeType = MimeType.valueOf("message/x.rsocket.authentication.bearer.v0")
val locations = requester.route("locate.radars.within")
.metadata(securityToken, mimeType)
.data(viewBox)
.retrieveFlow<AirportLocation>()
对于 Fire-and-Forget
,使用返回 Mono<Void>
的 send()
方法。请注意,Mono
仅表示消息已成功发送,而不表示已处理。
对于 Metadata-Push
,使用返回 Mono<Void>
的 sendMetadata()
方法。
注解响应者
RSocket 响应者可以实现为 @MessageMapping
和 @ConnectMapping
方法。@MessageMapping
方法处理单个请求,而 @ConnectMapping
方法处理连接级事件(设置和元数据推送)。注解响应者是对称支持的,用于从服务器端响应和从客户端响应。
服务器响应者
要在服务器端使用注解响应者,请将 RSocketMessageHandler
添加到您的 Spring 配置中,以检测带有 @MessageMapping
和 @ConnectMapping
方法的 @Controller
bean:
-
Java
-
Kotlin
@Configuration
static class ServerConfig {
@Bean
public RSocketMessageHandler rsocketMessageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.routeMatcher(new PathPatternRouteMatcher());
return handler;
}
}
@Configuration
class ServerConfig {
@Bean
fun rsocketMessageHandler() = RSocketMessageHandler().apply {
routeMatcher = PathPatternRouteMatcher()
}
}
然后通过 Java RSocket API 启动 RSocket 服务器,并按如下方式插入 RSocketMessageHandler
作为响应者:
-
Java
-
Kotlin
ApplicationContext context = ... ;
RSocketMessageHandler handler = context.getBean(RSocketMessageHandler.class);
CloseableChannel server =
RSocketServer.create(handler.responder())
.bind(TcpServerTransport.create("localhost", 7000))
.block();
import org.springframework.beans.factory.getBean
val context: ApplicationContext = ...
val handler = context.getBean<RSocketMessageHandler>()
val server = RSocketServer.create(handler.responder())
.bind(TcpServerTransport.create("localhost", 7000))
.awaitSingle()
RSocketMessageHandler
默认支持 复合 和 路由 元数据。如果您需要切换到不同的 mime 类型或注册其他元数据 mime 类型,可以设置其 rsocket-metadata-extractor。
您需要设置数据和元数据格式所需的 Encoder
和 Decoder
实例。您可能需要 spring-web
模块以获取编解码器实现。
默认情况下,SimpleRouteMatcher
用于通过 AntPathMatcher
匹配路由。我们建议插入 spring-web
中的 PathPatternRouteMatcher
以实现高效的路由匹配。RSocket 路由可以是分层的,但不是 URL 路径。这两个路由匹配器都默认配置为使用 "." 作为分隔符,并且不像 HTTP URL 那样进行 URL 解码。
RSocketMessageHandler
可以通过 RSocketStrategies
进行配置,如果您需要在同一个进程中的客户端和服务器之间共享配置,这可能会很有用:
-
Java
-
Kotlin
@Configuration
static class ServerConfig {
@Bean
public RSocketMessageHandler rsocketMessageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.setRSocketStrategies(rsocketStrategies());
return handler;
}
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
.encoders(encoders -> encoders.add(new Jackson2CborEncoder()))
.decoders(decoders -> decoders.add(new Jackson2CborDecoder()))
.routeMatcher(new PathPatternRouteMatcher())
.build();
}
}
@Configuration
class ServerConfig {
@Bean
fun rsocketMessageHandler() = RSocketMessageHandler().apply {
rSocketStrategies = rsocketStrategies()
}
@Bean
fun rsocketStrategies() = RSocketStrategies.builder()
.encoders { it.add(Jackson2CborEncoder()) }
.decoders { it.add(Jackson2CborDecoder()) }
.routeMatcher(PathPatternRouteMatcher())
.build()
}
客户端响应者
客户端的注解响应者需要在 RSocketRequester.Builder
中进行配置。有关详细信息,请参阅 rsocket-requester-client-responder。
@MessageMapping
一旦 rsocket-annot-responders-server 或 rsocket-annot-responders-client 响应者配置到位,就可以按如下方式使用 @MessageMapping
方法:
-
Java
-
Kotlin
@Controller
public class RadarsController {
@MessageMapping("locate.radars.within")
public Flux<AirportLocation> radars(MapRequest request) {
// ...
}
}
@Controller
class RadarsController {
@MessageMapping("locate.radars.within")
fun radars(request: MapRequest): Flow<AirportLocation> {
// ...
}
}
上述 @MessageMapping
方法响应路由为 "locate.radars.within" 的 Request-Stream 交互。它支持灵活的方法签名,并可选择使用以下方法参数:
方法参数 | 描述 |
---|---|
|
请求的有效载荷。这可以是具体值或异步类型(如 注意: 注解的使用是可选的。未指定简单类型且非其他受支持参数的方法参数被视为预期有效载荷。 |
|
用于向远程端发出请求的请求者。 |
|
从路由中根据映射模式中的变量提取的值,例如 |
|
注册用于提取的元数据值,如 rsocket-metadata-extractor 中所述。 |
|
注册用于提取的所有元数据值,如 rsocket-metadata-extractor 中所述。 |
返回值预期是一个或多个对象,将被序列化为响应负载。这可以是异步类型,如 Mono
或 Flux
,一个具体值,或者 void
或无值异步类型,如 Mono<Void>
。
@MessageMapping
方法支持的 RSocket 交互类型由输入(即 @Payload
参数)和输出的基数决定,其中基数表示以下内容:
基数 | 描述 |
---|---|
1 |
显式值,或单值异步类型,例如 |
Many |
多值异步类型,例如 |
0 |
对于输入,这意味着该方法没有 对于输出,这是 |
下表显示了所有输入和输出基数组合以及相应的交互类型:
输入基数 | 输出基数 | 交互类型 |
---|---|---|
0, 1 |
0 |
Fire-and-Forget, Request-Response |
0, 1 |
1 |
Request-Response |
0, 1 |
Many |
Request-Stream |
Many |
0, 1, Many |
Request-Channel |
@RSocketExchange
作为 @MessageMapping
的替代方案,您还可以使用 @RSocketExchange
方法处理请求。此类方法在 rsocket-interface 上声明,可以通过 RSocketServiceProxyFactory
用作请求者,或由响应者实现。
例如,作为响应者处理请求:
-
Java
-
Kotlin
public interface RadarsService {
@RSocketExchange("locate.radars.within")
Flux<AirportLocation> radars(MapRequest request);
}
@Controller
public class RadarsController implements RadarsService {
public Flux<AirportLocation> radars(MapRequest request) {
// ...
}
}
interface RadarsService {
@RSocketExchange("locate.radars.within")
fun radars(request: MapRequest): Flow<AirportLocation>
}
@Controller
class RadarsController : RadarsService {
override fun radars(request: MapRequest): Flow<AirportLocation> {
// ...
}
}
@RSocketExhange
和 @MessageMapping
之间存在一些差异,因为前者需要保持适用于请求者和响应者。例如,虽然 @MessageMapping
可以声明为处理任意数量的路由,并且每个路由都可以是模式,但 @RSocketExchange
必须声明为单个具体路由。在支持的方法参数方面也存在细微差异,请参阅 <<@MessageMapping,rsocket-annot-messagemapping>> 和 rsocket-interface 以获取支持参数的列表。
@RSocketExchange
可以在类型级别使用,以指定给定 RSocket 服务接口的所有路由的通用前缀。
@ConnectMapping
@ConnectMapping
处理 RSocket 连接开始时的 SETUP
帧,以及随后通过 METADATA_PUSH
帧进行的任何元数据推送通知,即 io.rsocket.RSocket
中的 metadataPush(Payload)
。
@ConnectMapping
方法支持与 <<@MessageMapping,rsocket-annot-messagemapping>> 相同的参数,但基于 SETUP
和 METADATA_PUSH
帧中的元数据和数据。@ConnectMapping
可以具有模式以将处理范围缩小到元数据中具有路由的特定连接,或者如果未声明任何模式,则所有连接都匹配。
@ConnectMapping
方法不能返回数据,并且必须声明为 void
或 Mono<Void>
作为返回值。如果处理对新连接返回错误,则连接将被拒绝。处理不得暂停以对连接的 RSocketRequester
发出请求。有关详细信息,请参阅 rsocket-requester-server。
MetadataExtractor
响应者必须解释元数据。 复合元数据 允许独立格式化的元数据值(例如,用于路由、安全性、跟踪),每个值都有自己的 mime 类型。应用程序需要一种方法来配置要支持的元数据 mime 类型,以及一种访问提取值的方法。
MetadataExtractor
是一个契约,用于获取序列化元数据并返回解码后的名称-值对,然后可以通过名称(例如通过注解处理程序方法中的 @Header
)访问这些名称-值对。
DefaultMetadataExtractor
可以给定 Decoder
实例来解码元数据。它开箱即用地内置支持 "message/x.rsocket.routing.v0",它将其解码为 String
并以 "route" 键保存。对于任何其他 mime 类型,您需要提供 Decoder
并按如下方式注册 mime 类型:
-
Java
-
Kotlin
DefaultMetadataExtractor extractor = new DefaultMetadataExtractor(metadataDecoders);
extractor.metadataToExtract(fooMimeType, Foo.class, "foo");
import org.springframework.messaging.rsocket.metadataToExtract
val extractor = DefaultMetadataExtractor(metadataDecoders)
extractor.metadataToExtract<Foo>(fooMimeType, "foo")
复合元数据非常适合组合独立的元数据值。但是,请求者可能不支持复合元数据,或者可能选择不使用它。为此,DefaultMetadataExtractor
可能需要自定义逻辑将解码后的值映射到输出映射。以下是使用 JSON 作为元数据的示例:
-
Java
-
Kotlin
DefaultMetadataExtractor extractor = new DefaultMetadataExtractor(metadataDecoders);
extractor.metadataToExtract(
MimeType.valueOf("application/vnd.myapp.metadata+json"),
new ParameterizedTypeReference<Map<String,String>>() {},
(jsonMap, outputMap) -> {
outputMap.putAll(jsonMap);
});
import org.springframework.messaging.rsocket.metadataToExtract
val extractor = DefaultMetadataExtractor(metadataDecoders)
extractor.metadataToExtract<Map<String, String>>(MimeType.valueOf("application/vnd.myapp.metadata+json")) { jsonMap, outputMap ->
outputMap.putAll(jsonMap)
}
通过 RSocketStrategies
配置 MetadataExtractor
时,您可以让 RSocketStrategies.Builder
使用配置的解码器创建提取器,然后简单地使用回调来自定义注册,如下所示:
-
Java
-
Kotlin
RSocketStrategies strategies = RSocketStrategies.builder()
.metadataExtractorRegistry(registry -> {
registry.metadataToExtract(fooMimeType, Foo.class, "foo");
// ...
})
.build();
import org.springframework.messaging.rsocket.metadataToExtract
val strategies = RSocketStrategies.builder()
.metadataExtractorRegistry { registry: MetadataExtractorRegistry ->
registry.metadataToExtract<Foo>(fooMimeType, "foo")
// ...
}
.build()
RSocket 接口
Spring Framework 允许您将 RSocket 服务定义为带有 @RSocketExchange
方法的 Java 接口。您可以将此类接口传递给 RSocketServiceProxyFactory
以创建代理,该代理通过 rsocket-requester 执行请求。您还可以将该接口实现为处理请求的响应者。
首先创建带有 @RSocketExchange
方法的接口:
interface RadarService {
@RSocketExchange("radars")
Flux<AirportLocation> getRadars(@Payload MapRequest request);
// more RSocket exchange methods...
}
现在您可以创建一个代理,当调用方法时执行请求:
RSocketRequester requester = ... ;
RSocketServiceProxyFactory factory = RSocketServiceProxyFactory.builder(requester).build();
RadarService service = factory.createClient(RadarService.class);
您还可以实现该接口以作为响应者处理请求。请参阅 rsocket-annot-rsocketexchange。
方法参数
带注解的 RSocket 交换方法支持灵活的方法签名,具有以下方法参数:
方法参数 | 描述 |
---|---|
|
添加路由变量以传递给 |
|
设置请求的输入有效载荷。这可以是具体值,或任何可以通过 |
|
输入有效载荷中元数据条目的值。这可以是任何 |
|
元数据条目的 |
返回值
带注解的 RSocket 交换方法支持返回值,这些返回值可以是具体值,也可以是任何可以通过 ReactiveAdapterRegistry
适配为 Reactive Streams Publisher
的值生产者。
默认情况下,RSocket 服务方法的同步(阻塞)方法签名的行为取决于底层 RSocket ClientTransport
的响应超时设置以及 RSocket keep-alive 设置。RSocketServiceProxyFactory.Builder
确实公开了一个 blockTimeout
选项,也允许您配置阻塞响应的最长时间,但我们建议在 RSocket 级别配置超时值以获得更多控制。