整合RSocket实时数据通信

大家好，欢迎来到IT知识分享网。

环境：SpringBoot2.7.12

1. 概述

RSocket 是一种二进制协议，可用于 TCP、WebSockets 和 Aeron 等字节流传输的应用协议，具有以下交互模型：

• Request-Response: 发送一条信息，接收一条信息。
• Request-Stream: 发送一条消息并接收返回的消息流。
• Channel: 双向发送消息流。
• Fire-and-Forget: 发送单向消息。

一旦建立了初始连接，“客户端”与“服务器”之间的区别就消失了，因为两端变得对称，每一端都可以发起上述交互之一。这就是为什么在协议中称参与端的为“请求者”和“响应者”，而上述交互称为“请求流”或简单地称为“请求”。

RSocket 协议的主要特点和优势：

• 跨网络边界的响应式流语义——对于流请求，如请求流和通道，背压信号在请求者和响应者之间传递，允许请求者在源端减慢响应者的速度，从而减少对网络层拥塞控制的依赖，以及在网络层或任何层缓冲的需要。
• 请求节流(Request throttling)——这个特性被命名为“租赁”(LEASE)，以从两端发送的租赁帧命名，用于限制给定时间内另一端允许的请求总数。租约是定期更新的。
• 会话恢复——这是为连接丢失而设计的，需要维护一些状态。状态管理对应用程序是透明的，并与背压相结合，可以在可能的情况下停止生产者并减少所需的状态量。
• 大消息的碎片化和重组。
• 保活（心跳）。

在Java中基于 Project Reactor 和用于传输的 Reactor Netty 构建。

2. 协议简介

Connecting

最初，客户端通过 TCP 或 WebSocket 等低级流传输方式连接服务器，并向服务器发送 SETUP 帧，以设置连接参数。服务器可能会拒绝 SETUP 帧，但一般情况下，在发送（客户端）和接收（服务器）SETUP 帧后，双方都可以开始发出请求，除非 SETUP 表示使用租用语义来限制请求数量，在这种情况下，双方都必须等待另一端发出 LEASE 帧才能允许发出请求。

发出请求(Making Requests)

建立连接后，双方可通过 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 类型都在设置框架中声明，并适用于给定连接上的所有请求。

虽然所有报文都可以包含元数据，但路由等元数据通常是按请求提供的，因此只包含在请求的第一条报文中，即 REQUEST_RESPONSE、REQUEST_STREAM、REQUEST_CHANNEL 或 REQUEST_FNF 框架之一。

3. 应用场景

1. 游戏开发：游戏需要实时、低延迟的通信，以提供更好的用户体验。RSocket提供了一种可靠、高效的通信方式，可以在不同的游戏组件之间进行通信，例如服务器和客户端之间的通信，或者客户端与客户端之间的通信。
2. 实时流媒体：实时流媒体应用需要将大量的数据从服务器传输到客户端，并且需要保证数据的实时性和稳定性。RSocket提供了一种可靠的数据传输方式，可以保证数据的有序性和完整性。
3. 物联网（IoT） ：物联网设备需要通过网络进行通信，以实现设备的远程控制和数据采集。RSocket可以用于建立可靠、高效的连接，使得设备可以快速地传输和接收数据。
4. 分布式系统：分布式系统需要将不同的组件连接在一起，以实现协同工作。RSocket提供了一种可靠、高效的通信方式，可以用于在不同的组件之间进行通信。
5. 实时数据分析：实时数据分析需要对大量的数据进行实时处理和分析。RSocket可以用于将数据从数据源传输到处理节点，并保证数据的实时性和完整性。

4. 实战案例

案例会分别演示RSocket的4种交换模式。

依赖配置

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-rsocket</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency> 

新建两个项目，如下：

服务端配置

spring: rsocket: server: port: 9898 transport: tcp 

客户端消息处理模板类

@Service public class PackRSocketService { private final RSocketRequester rsocketRequester; public PackRSocketService(RSocketRequester.Builder rsocketRequesterBuilder) { this.rsocketRequester = rsocketRequesterBuilder.tcp("localhost", 9898) ; } } 

接下来就会基于上面的配置进行各种交互模式的演示。

4.1 Request-Response模式

服务端

@MessageMapping("message") public Mono<String> handleMessage(Mono<String> message) { return message.doOnNext(msg -> { System.out.printf("接收到消息：%s%n", msg) ; }).map(msg -> "服务器成功收到了你的消息！！！") ; } 

客户端

// Request-Response 发送一条信息，接收一条信息。 public void sendMessage(String body) { this.rsocketRequester .route("message") .data(body) .retrieveMono(String.class) .subscribe(System.out::println) ; } 

消息发送

@GetMapping("/message") public Object message() { pss.sendMessage(String.valueOf(System.nanoTime())) ; return "message" ; } 

运行结果

4.2 Request-Stream模式

服务端

// 必须返回Flux @MessageMapping("stream") public Flux<String> handleStream() { return Flux .interval(Duration.ofSeconds(2)) // 随机生成 .map(i -> String.valueOf(new Random().nextInt())) // 只在此通道中获取10个值 .take(10) .doOnComplete(() -> { System.out.println("completed...") ; }) ; } 

客户端

// Request-Stream 发送一条消息并接收返回的消息流。 public void sendStream() { this.rsocketRequester .route("stream") .retrieveFlux(String.class) .subscribe(ret -> { System.out.printf("%s - 接受到数据: %s%n", Thread.currentThread().getName(), ret) ; }) ; } 

运行结果

4.3 Channel

服务端

@MessageMapping("channel") public Flux<String> handleChannel(Flux<String> datas) { return datas.doOnNext(ret -> { System.out.printf("【server】%s - 接收到数据: %s%n", Thread.currentThread().getName(), ret) ; }).map(ret -> { return ret + " - " + new Random().nextInt(1000) ; }) ; } 

客户端

// Channel 双向发送消息流。 public void sendChannel() { this.rsocketRequester .route("channel") .data(Flux.just("1", "2", "3", "4", "5", "6").delayElements(Duration.ofSeconds(1))) .retrieveFlux(String.class) .subscribe(ret -> { System.out.printf("【client】%s - 接受到数据: %s%n", Thread.currentThread().getName(), ret) ; }) ; } 

运行结果

4.4 Fire-and-Forget

服务端

@MessageMapping("faf") public Mono<Void> handleFireAndForget(Mono<String> data) { return data.doOnNext(ret -> { System.out.printf("【server】%s - 接收到数据: %s%n", Thread.currentThread().getName(), ret) ; }).then() ; } 

客户端

// Fire-and-Forget 发送单向消息。 public void sendFireAndForget() { this.rsocketRequester .route("faf") .data(Mono.just(String.valueOf(new Random().nextInt(1000)))) .send() .subscribe() ; } 

运行结果

总结：选择RSocket还是HTTP作为微服务间通信方式需要根据具体的业务场景和需求进行综合考虑。如果对性能、低延迟和双向通信有较高要求，RSocket可能更适合；如果对成熟度、简单性和跨平台性，HTTP可能更适合。