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

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

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;
}

协议层

Servlet	基于HTTP协议的，HTTP并非非常简单，1.1，2.0版本开始是有点复杂的
RSocket	自定义二进制协议，RSocket定位高性能通讯，比HTTP高非常多（号称10倍）

通讯模式

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

message

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

RSocket && dubbo

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

RSocket && Spring

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

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 {@Overridepublic Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) {return Mono.just(new AbstractRSocket() {@Overridepublic Mono<Void> fireAndForget(Payload payload) {//System.out.println(payload.getDataUtf8());payload.release();return Mono.empty();}@Overridepublic Mono<Payload> requestResponse(Payload payload) {return Mono.just(payload);}@Overridepublic 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;@Overridepublic 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 {@Overridepublic Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) {return Mono.just(new AbstractRSocket() {@Overridepublic 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订阅的实时查询。