响应式编程之网络新约：RSocket

水深无声 2022-04-25 00:44 203阅读 0赞

响应式reactive是Java中高效应用的下一个前沿，但它目前主要有两个障碍：数据访问和网络。RSocket是一种新的第7层语言无关的应用网络协议（解决后者），它由Facebook，Netifi和Pivotal等工程师开发，提供Java，JavaScript，C ++和Kotlin等实现，RSocket与Servlet并不是同类的产品。

### RSocket ###

[RSocket ][RSocket]RSocket是一个二进制的协议，以异步消息的方式提供4种对等的交互模型，以字节流的方式运行在TCP, WebSockets, Aeron等传输层之上。RSocket专门设计用于与Reactive风格应用配合使用，这些应用程序基本上是非阻塞的，并且通常（但不总是）与异步行为配对。它是传输无关的，支持 TCP、[WebSocket][]和[Aeron UDP][]协议，并支持无语义损失的混合传输协议——回压和流量控制仍然有效。

它还支持连接恢复。当你建立 RSocket 连接时，你可以指定前一个连接的 ID，如果流仍然在服务器的内存中，则你可以继续消费你的流。

Payload就是前面说到基于消息通讯，那就是拿到消息返回消息。对于一个消息来说，由两部分组成，原信息(metadata)和数据(data)。Mono和Flux是用来处理异步的关键字，这是Reactive编程要求。

public interface RSocket extends Availability, Closeable {
        /** 
        * 推送元信息，数据可以自己定
        */
        Mono<Void> metadataPush(Payload payload);
        /**请求/响应
        * 当你发送一个请求并接收一个响应时，该协议也比 HTTP 更具优势，因为它是异步且多路复用的
        */
        Mono<Payload> requestResponse(Payload payload);
        /**即发即忘
        * 请求/响应的优化，在不需要响应时非常有用,比如用于非关键事件的日志记录
        */
        Mono<Void> fireAndForget(Payload payload);
        /**请求/流
        * 类似于返回集合的请求/响应，集合将以流的方式返回，而不是等到查询完成，例如，发送一个银行帐号，使用一个实时的帐户事务流进行响应
        */
        Flux<Payload> requestStream(Payload payload);
        /**通道
        * 允许任意交互模型的双向消息流
        */
        Flux<Payload> requestChannel(Publisher<Payload> payloads);
        /**健康度检查
        * double值可以作为权重，如1.0表示处理能力非常好，0.8一般
        */
        default double availability() {
            return isDisposed() ? 0.0 : 1.0;
        }
    }

### RSocket vs Servlet ###

> Servlet是一套Java的API规范，基于HTTP协议之上。主要功能提供HTTP服务的class，就是通过HTTP Request，处理后，最终调用HTTP Response完成输出！

public abstract class HttpServlet extends Servlet {
        protected abstract void doGet(HttpServletRequest request,HttpServletResponse response)  throws ServletException, IOException;
        protected abstract void doPost(HttpServletRequest request,HttpServletResponse response)  throws ServletException, IOException;
    }

**协议层**

<table> 
 <tbody> 
  <tr> 
   <td style="width:140px;">Servlet</td> 
   <td>基于HTTP协议的，HTTP并非非常简单，1.1，2.0版本开始是有点复杂的</td> 
  </tr> 
  <tr> 
   <td style="width:140px;">RSocket</td> 
   <td>自定义二进制协议，RSocket定位高性能通讯，比HTTP高非常多（号称10倍）</td> 
  </tr> 
 </tbody> 
</table>

**通讯模式**

<table> 
 <tbody> 
  <tr> 
   <td style="width:140px;">Servlet</td> 
   <td>都是request/response模式，所以也叫做 request command，其他例如流式推送、fireAndForget和双向通讯，Servlet2.0都不支持，但是这些指令都是为浏览器设计的，并非为服务通讯设计的</td> 
  </tr> 
  <tr> 
   <td style="width:140px;">RSocket</td> 
   <td>对等通讯，不再介于传统的理解是Client -&gt; Server模式，RSocket没有这个概念，大家的地位是对等的，都可以在server端，我调用你的服务，你也可以调用我的服务</td> 
  </tr> 
 </tbody> 
</table>

![20190430101358828.png][]

**message**

<table> 
 <tbody> 
  <tr> 
   <td style="width:140px;">Servlet</td> 
   <td>HTTP1.1是基于文本的通讯，2.0是基于message的（二进制），基于message的好处是异步化。message都必须有一个ID，这个消息发送出去后，就不用等立即返回，可以继续发其他message，收到message后，再根据返回的message ID和之前的发出去的message ID进行匹配。</td> 
  </tr> 
  <tr> 
   <td style="width:140px;">RSocket</td> 
   <td>基于message</td> 
  </tr> 
 </tbody> 
</table>

### RSocket && dubbo ###

Dubbo 在 [3.0.0-SNAPSHOT][] 版本里基于 RSocket 对响应式编程提供了支持，用户可以非常方便的使用RSocket的语法。使用实例可以[参阅官方][Link 1]，待正式版发布后，接触RSocket的机会也会越来越多。

### RSocket &&  Spring ###

随着Spring Cloud的推出，Spring Framework 5.2 即将要把RSocket作为缺省的通讯协议，springBoot中[提供相应支持][Link 2]。

### RSocket && 微服务 ###

RSocket的主要障碍是应用程序之间必须要用RSocket通讯。微服务普及后，其为了“简化”微服务之间的通讯，引入了很多层的技术栈。这当然是好事，但是很多的决定是由于收到上一代的通讯协议的技术所限制。

### 示例 ###

spring-boot-starter-rsocket其实也已经封装好了，使用起来比下面例子更加简单方便，感觉离rpc更近了一步

public final class ChannelEchoClient {
      static final Payload payload1 = ByteBufPayload.create("Hello ");
      public static void main(String[] args) {
        RSocketFactory.receive()
            .frameDecoder(PayloadDecoder.ZERO_COPY)
            .acceptor(new SocketAcceptorImpl())
            .transport(LocalServerTransport.create("localhost"))
            .start()
            .subscribe();
    
        RSocket socket =
            RSocketFactory.connect()
                .keepAliveAckTimeout(Duration.ofMinutes(10))
                .frameDecoder(PayloadDecoder.ZERO_COPY)
                .transport(LocalClientTransport.create("localhost"))
                .start()
                .block();
    
        Flux.range(0, 100000000)
            .concatMap(i -> socket.fireAndForget(payload1.retain()))
            .blockLast();
      }
      private static class SocketAcceptorImpl implements SocketAcceptor {
        @Override
        public Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) {
          return Mono.just(
              new AbstractRSocket() {
    
                @Override
                public Mono<Void> fireAndForget(Payload payload) {
                  //System.out.println(payload.getDataUtf8());
                  payload.release();
                  return Mono.empty();
                }
    
                @Override
                public Mono<Payload> requestResponse(Payload payload) {
                  return Mono.just(payload);
                }
    
                @Override
                public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
                  return Flux.from(payloads).subscribeOn(Schedulers.single());
                }
              });
        }
      }
    }
    //request/response
    public final class HelloWorldClient {
      public static void main(String[] args) {
        RSocketFactory.receive()
            .acceptor(
                (setupPayload, reactiveSocket) ->
                    Mono.just(
                        new AbstractRSocket() {
                          boolean fail = true;
    
                          @Override
                          public Mono<Payload> requestResponse(Payload p) {
                            if (fail) {
                              fail = false;
                              return Mono.error(new Throwable());
                            } else {
                              return Mono.just(p);
                            }
                          }
                        }))
            .transport(TcpServerTransport.create("localhost", 7000))
            .start()
            .subscribe();
    
        RSocket socket =
            RSocketFactory.connect()
                .transport(TcpClientTransport.create("localhost", 7000))
                .start()
                .block();
    
        socket
            .requestResponse(DefaultPayload.create("Hello"))
            .map(Payload::getDataUtf8)
            .onErrorReturn("error")
            .doOnNext(System.out::println)
            .block();
    
        socket
            .requestResponse(DefaultPayload.create("Hello"))
            .map(Payload::getDataUtf8)
            .onErrorReturn("error")
            .doOnNext(System.out::println)
            .block();
    
        socket
            .requestResponse(DefaultPayload.create("Hello"))
            .map(Payload::getDataUtf8)
            .onErrorReturn("error")
            .doOnNext(System.out::println)
            .block();
    
        socket.dispose();
      }
    }
    //request/stream
    public final class StreamingClient {
      public static void main(String[] args) {
        RSocketFactory.receive()
            .acceptor(new SocketAcceptorImpl())
            .transport(TcpServerTransport.create("localhost", 7000))
            .start()
            .subscribe();
    
        RSocket socket =
            RSocketFactory.connect()
                .transport(TcpClientTransport.create("localhost", 7000))
                .start()
                .block();
    
        socket
            .requestStream(DefaultPayload.create("Hello"))
            .map(Payload::getDataUtf8)
            .doOnNext(System.out::println)
            .take(10)
            .then()
            .doFinally(signalType -> socket.dispose())
            .then()
            .block();
      }
      private static class SocketAcceptorImpl implements SocketAcceptor {
        @Override
        public Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) {
          return Mono.just(
              new AbstractRSocket() {
                @Override
                public Flux<Payload> requestStream(Payload payload) {
                  return Flux.interval(Duration.ofMillis(100))
                      .map(aLong -> DefaultPayload.create("Interval: " + aLong));
                }
              });
        }
      }
    }

### 参阅资料 ###

*  [Introduction to RSocket][]

REST的最大限制是它与HTTP相关联，经常使用REST的原因是它易于调试，因为它是“人类可读”。开源RSocket专为服务而设计。它是一种面向连接的消息驱动协议，在应用程序级别具有内置流控制。它既可以在浏览器中同样使用，也可以在服务器上使用。这意味着您可以流式传输数据或执行Pub / Sub而无需设置应用程序队列。在Facebook，RSocket用于名为LiveServer的服务，该服务负责响应可被视为GraphQL订阅的实时查询。![89576152][]

[RSocket]: http://rsocket.io/
[WebSocket]: https://en.wikipedia.org/wiki/WebSocket
[Aeron UDP]: https://github.com/real-logic/aeron
[20190430101358828.png]: /images/20220210/644d99eccd134aba88afbd16081f6c70.png
[3.0.0-SNAPSHOT]: https://github.com/apache/incubator-Dubbo/tree/3.x-dev
[Link 1]: https://github.com/apache/incubator-dubbo-samples/tree/3.x/dubbo-samples-rsocket
[Link 2]: https://github.com/spring-projects/spring-boot/tree/master/spring-boot-project/spring-boot-starters/spring-boot-starter-rsocket
[Introduction to RSocket]: https://www.baeldung.com/rsocket
[89576152]: /images/20220210/7710ffe2e5bd47d2aa09706ac122b179.png