序
本文主要研究一下 rsocket-java
RSocket
rsocket-core-0.12.1-sources.jar!/io/rsocket/RSocket.java
public interface RSocket extends Availability, Closeable {
/**
* Fire and Forget interaction model of {@code RSocket}.
*
* @param payload Request payload.
* @return {@code Publisher} that completes when the passed {@code payload} is successfully
* handled, otherwise errors.
*/
Mono<Void> fireAndForget(Payload payload);
/**
* Request-Response interaction model of {@code RSocket}.
*
* @param payload Request payload.
* @return {@code Publisher} containing at most a single {@code Payload} representing the
* response.
*/
Mono<Payload> requestResponse(Payload payload);
/**
* Request-Stream interaction model of {@code RSocket}.
*
* @param payload Request payload.
* @return {@code Publisher} containing the stream of {@code Payload}s representing the response.
*/
Flux<Payload> requestStream(Payload payload);
/**
* Request-Channel interaction model of {@code RSocket}.
*
* @param payloads Stream of request payloads.
* @return Stream of response payloads.
*/
Flux<Payload> requestChannel(Publisher<Payload> payloads);
/**
* Metadata-Push interaction model of {@code RSocket}.
*
* @param payload Request payloads.
* @return {@code Publisher} that completes when the passed {@code payload} is successfully
* handled, otherwise errors.
*/
Mono<Void> metadataPush(Payload payload);
@Override
default double availability() {
return isDisposed() ? 0.0 : 1.0;
}
}
RSocket 接口继承了 Availability(定义 double availability() 方法 ) 及 Closeable(定义了 Mono<Void> onClose() 方法 ) 接口
RSocket 定义了 fireAndForget、requestResponse、requestStream、requestChannel 方法分别对应 4 种 Interaction Model
RSocket 的 Frame 包含 metadata 及 data payload，其中 metadata 可选，可以用于描述 data payload，因而 RSocket 还定义了 metadataPush 方法用于 push metadata
Interaction Model
fireAndForget
@Test
public void testFireAndForget() throws InterruptedException {
//SERVER
RSocketFactory.receive()
.acceptor(
(setupPayload, reactiveSocket) ->
Mono.just(
new AbstractRSocket() {
@Override
public Mono<Void> fireAndForget(Payload payload) {
System.out.printf(“fire-forget: %s%n”, payload.getDataUtf8());
return Mono.empty();
}
}))
.transport(TcpServerTransport.create(“localhost”, 8080))
.start()
.subscribe();
//CLIENT
RSocket socket =
RSocketFactory.connect()
.transport(TcpClientTransport.create(“localhost”, 8080))
.start()
.block();
socket
.fireAndForget(DefaultPayload.create(“Hello”))
.block();
socket.dispose();
TimeUnit.SECONDS.sleep(5);
}
类似 udp，无需 ack，比较适合 metrics 上报、访问日志上报等
requestResponse
@Test
public void testRequestResponse(){
//SERVER
RSocketFactory.receive()
.acceptor(
(setupPayload, reactiveSocket) ->
Mono.just(
new AbstractRSocket() {
@Override
public Mono<Payload> requestResponse(Payload p) {
return Mono.just(p);
}
}))
.transport(TcpServerTransport.create(“localhost”, 8080))
.start()
.subscribe();
//CLIENT
RSocket socket =
RSocketFactory.connect()
.transport(TcpClientTransport.create(“localhost”, 8080))
.start()
.block();
socket
.requestResponse(DefaultPayload.create(“Hello”))
.map(Payload::getDataUtf8)
.onErrorReturn(“error”)
.doOnNext(System.out::println)
.block();
socket.dispose();
}
类似 http，但是优于 http，因为它是异步的，而且是 multiplexed
requestStream
@Test
public void testRequestStream(){
//SERVER
RSocketFactory.receive()
.acceptor(new SocketAcceptor() {
@Override
public Mono<RSocket> accept(ConnectionSetupPayload connectionSetupPayload, RSocket rSocket) {
return Mono.just(
new AbstractRSocket() {
@Override
public Flux<Payload> requestStream(Payload payload) {
return Flux.interval(Duration.ofMillis(100))
.map(aLong -> DefaultPayload.create(“Interval: ” + aLong));
}
});
}
})
.transport(TcpServerTransport.create(“localhost”, 7000))
.start()
.subscribe();
//CLIENT
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();
}
类似 Request-Response(返回 Mono)，只不过返回的是 Flux
requestChannel
@Test
public void testRequestChannel(){
//SERVER
RSocketFactory.receive()
.acceptor(new SocketAcceptor(){
@Override
public Mono<RSocket> accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
return Mono.just(
new AbstractRSocket() {
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
return Flux.from(payloads)
.map(Payload::getDataUtf8)
.map(s -> “Echo: ” + s)
.map(DefaultPayload::create);
}
});
}
})
.transport(TcpServerTransport.create(“localhost”, 8080))
.start()
.subscribe();
//CLIENT
RSocket socket =
RSocketFactory.connect()
.transport(TcpClientTransport.create(“localhost”, 8080))
.start()
.block();
socket
.requestChannel(
Flux.interval(Duration.ofMillis(1000)).map(i -> DefaultPayload.create(“Hello”)))
.map(Payload::getDataUtf8)
.doOnNext(System.out::println)
.take(10)
.doFinally(signalType -> socket.dispose())
.then()
.block();
}
类似 websocket，可以双向通信
MetadataPush
@Test
public void testMetadataPush() throws InterruptedException {
//SERVER
RSocketFactory.receive()
.acceptor(
(setupPayload, reactiveSocket) ->
Mono.just(
new AbstractRSocket() {
@Override
public Mono<Void> metadataPush(Payload payload) {
System.out.printf(“metadataPush: %s%n”, payload.getDataUtf8());
return Mono.empty();
}
}))
.transport(TcpServerTransport.create(“localhost”, 8080))
.start()
.subscribe();
//CLIENT
RSocket socket =
RSocketFactory.connect()
.transport(TcpClientTransport.create(“localhost”, 8080))
.start()
.block();
socket
.metadataPush(DefaultPayload.create(“hello”,”version=1.0.0+”))
.block();
socket.dispose();
TimeUnit.SECONDS.sleep(5);
}
RSocket 还定义了 metadataPush 方法，与 fireAndForget 方法不同的是 metadataPush 方法会等待 data pushed 成功，然后在接收到对方发送的 complete signal 时 complete
小结
RSocket 是一种 bi-directional、multiplexed、message-based 的二进制协议
RSocket 有四种 Interaction Model，分别是 Request-Response、Fire-and-Forget、Request-Stream、Channel
RSocket 的 Frame 包含 metadata 及 data payload，其中 metadata 可选，可以用于描述 data payload；除了可以在 4 种 Interaction Model 对应方法的 Payload 参数中设置 metadata 外，还可以使用 RSocket 定义的 metadataPush 方法来专门 push metadata
doc
rsocket.io