Spring Boot与RSocket实现高效实时数据通信

作者介绍：✌️大厂全栈码农|毕设实战开发，专注于大学生项目实战开发、讲解和毕业答疑辅导。

推荐订阅精彩专栏 👇🏻 避免错过下次更新

Springboot项目精选实战案例

更多项目：CSDN主页YAML墨韵

学如逆水行舟，不进则退。学习如赶路，不能慢一步。

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. 应用场景

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