RSocket协议初识（一）

祖麻雀

2023-12-01

前言

前几天无聊翻SpringBoot官方文档，无意中发现文档中增加了一个章节叫RSocket协议的鬼东西，遂研究了一下。

RSocket是什么？

RSocket是一种二进制字节流传输协议，位于OSI模型中的5~6层，底层可以依赖TCP、WebSocket、Aeron协议。

RSocket设计目标是什么？

1、支持对象传输，包括request\response、request\stream、fire and forget、channel
2、支持应用层流量控制
3、支持单连接双向、多次复用
4、支持连接修复
5、更好的使用WebSocket和Aeron协议

RSocket与其他协议有什么区别？

对比Http1.x

Http1.x只支持request\response，但是现实应用中并不是所有请求都需要有回应（Fire And Forget）、有的需求需要一个请求返回一个数据流（request\stream）、有的还需要双向数据传输(channel)。

对比Http2.x

http2.x不支持应用层流量控制、伪双向传输，即服务端push数据本质上还是对客户端请求的响应，而不是直接推送。RSocket做到了真正的双向传输，使得服务端可以调用客户端服务，使得服务端和客户端在角色上完全对等，即两边同时是Requester和Responder。

对比grpc

grpc需要依赖protobuf，本质上还是http2.x。RSocket不限制编解码，可以是json、protobuf等等。
性能上grpc要差一些：详见压测对比，https://dzone.com/articles/rsocket-vs-grpc-benchmark

对比TCP

其实两者不在一个层面，为啥要作比较呢，因为netty让tcp层的编程也很容易，但是需要自定义传输协议，比如定义header、body长度等等，用起来还是很麻烦的。

对比WebSocket

websocket不支持应用层流量控制，本质上也是一端请求另一端响应，不支持连接修复。

RSocket协议的形式是什么？

连接上传输的数据是流（Stream）
流（Stream）由帧（Frame）组成
帧（Frame）包含了元数据（MetaData）与业务数据（Data）

结论：

基于RSocket协议，我们的业务数据会被打包成帧，并以帧流的形式在客户端与服务端互相传输。所以RSocket的所有特性都是基于这个帧流实现的。后续有时间会针对每个帧类型做解析。

RSocket适用于哪些场景？

1、移动设备与服务器的连接。

数据双向传输，且支持流量控制。支持背压，背压的意思：如果客户端请求服务端过快，那么服务端会堆积请求，最终耗光资源。有了背压服务端可以根据自己的资源来控制客户端的请求速度，即调用客户端告诉他别发那么快。
支持连接修复，比如手机进地铁之后，网络断开一段时间，其他协议需要重新建立连接，RSocket则可以修复连接继续传输帧数据。

2、微服务场景。

spring cloud目前支持的http协议，不能fire and forget、不能请求流数据、不能单连接双向调用；替换成RSocket之后可以满足以上需求的同时提高性能。且针对服务治理、负载均衡等RSocket都在慢慢完善。

3、由于微服务和移动设备的普及，RSocket火起来应该就是这几年的事儿。

BB了这么多你给我上个代码

SpringBoot中的使用

step1、构建SpringBoot项目，引入依赖

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

step2、编写需要传输的消息类和服务器类


import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.time.Instant;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Message {
    private String from;
    private String to;
    private long index;
    private long created = Instant.now().getEpochSecond();

    public Message(String from, String to) {
        this.from = from;
        this.to = to;
        this.index = 0;
    }

    public Message(String from, String to, long index) {
        this.from = from;
        this.to = to;
        this.index = index;
    }
}

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.ConnectMapping;
import org.springframework.stereotype.Controller;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

@Slf4j
@Controller
public class RSocketController {

    private final List<RSocketRequester> CLIENTS = new ArrayList<>();

    @MessageMapping("request-response")
    public Message requestResponse(Message request) {
        log.info("收到请求: {}", request);
        return new Message("服务端", "客户端");
    }

    @MessageMapping("fire-and-forget")
    public void fireAndForget(Message request) {
        log.info("收到fire-and-forget请求: {}", request);
    }

    @MessageMapping("stream")
    Flux<Message> stream(Message request) {
        log.info("收到流式请求: {}", request);
        return Flux
                .interval(Duration.ofSeconds(1))
                .map(index -> new Message(”服务端“, "客户端", index))
                .log();
    }

    @MessageMapping("channel")
    Flux<Message> channel(final Flux<Duration> settings) {
        return settings
                .doOnNext(setting -> log.info("发射间隔为 {} 秒.", setting.getSeconds()))
                .switchMap(setting -> Flux.interval(setting)
                        .map(index -> new Message("服务端", "客户端", index)))
                .log();
    }
}

step3、配置文件里增加配置项

spring.main.lazy-initialization=true
spring.rsocket.server.port=7000

step4、编写客户端代码


import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.time.Instant;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Message {
    private String from;
    private String to;
    private long index;
    private long created = Instant.now().getEpochSecond();

    public Message(String from, String to) {
        this.from = from;
        this.to = to;
        this.index = 0;
    }

    public Message(String from, String to, long index) {
        this.from = from;
        this.to = to;
        this.index = index;
    }
}

import java.time.Duration;
import javax.annotation.PreDestroy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import lombok.extern.slf4j.Slf4j;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Slf4j
@RestController
public class RSocketClient {

    private final RSocketRequester rsocketRequester;
    private static Disposable disposable;

    @Autowired
    public RSocketClient(RSocketRequester.Builder rsocketRequesterBuilder, RSocketStrategies strategies) {
        this.rsocketRequester = rsocketRequesterBuilder
                .rsocketStrategies(strategies)
                .connectTcp("localhost", 7000)
                .block();

        this.rsocketRequester.rsocket()
                .onClose()
                .doOnError(error -> log.warn("发生错误，链接关闭"))
                .doFinally(consumer -> log.info("链接关闭"))
                .subscribe();
    }

    @PreDestroy
    void shutdown() {
        rsocketRequester.rsocket().dispose();
    }

    @GetMapping("request-response")
    public Message requestResponse() {
        Message message = this.rsocketRequester
                .route("request-response")
                .data(new Message("客户端", "服务器"))
                .retrieveMono(Message.class)
                .block();
        log.info("客户端request-response收到响应 {}", message);
        return message;
    }

    @GetMapping("fire-and-forget")
    public String fireAndForget() {
        this.rsocketRequester
                .route("fire-and-forget")
                .data(new Message("客户端", "服务器"))
                .send()
                .block();
        return "fire and forget";
    }

    @GetMapping("stream")
    public String stream() {
        disposable = this.rsocketRequester
                .route("stream")
                .data(new Message("客户端", "服务器"))
                .retrieveFlux(Message.class)
                .subscribe(message -> log.info("客户端stream收到响应 {}", message));
        return "stream";
    }

    @GetMapping("channel")
    public String channel() {
        Mono<Duration> setting1 = Mono.just(Duration.ofSeconds(1));
        Mono<Duration> setting2 = Mono.just(Duration.ofSeconds(3)).delayElement(Duration.ofSeconds(5));
        Mono<Duration> setting3 = Mono.just(Duration.ofSeconds(5)).delayElement(Duration.ofSeconds(15));
        Flux<Duration> settings = Flux.concat(setting1, setting2, setting3)
                .doOnNext(d -> log.info("客户端channel发送消息 {}", d.getSeconds()));
        disposable = this.rsocketRequester
                .route("channel")
                .data(settings)
                .retrieveFlux(Message.class)
                .subscribe(message -> log.info("客户端channel收到响应 {}", message));
        return "channel";
    }

}

step5、启动服务端、启动客户端，打开浏览器访问localhost:8080/fire-and-forget等测试效果

代码解析

@MessageMapping：Spring提供的注解，用于路由，与@GetMapping等功能类似
Mono：响应式编程里用于返回0-1个结果
Flux：响应式编程里用于返回0-N个结果
Disposable：断流器，为true的时候两边不能传输数据

What Next？

协议原理解析
由于RSocket社区还不够活跃，Git上的代码也是刚刚起步，还在不断更新中，相关功能也在不断完善中，后续随着官方新内容的更新我也会跟着更新。
RSocket中很多概念如Mono、Flux、Disposable、背压、流式处理等都是响应式编程中的概念，想了解响应式编程可以查看：http://reactivex.io/ 中的文档，其中包括了RXJava等RX系列的各种语言的Demo。