前言
在RSocket协议初识（一）中我们使用SpringBoot创建了简单的RSocket通讯模型，本篇我们将介绍纯Java使用Demo。本文代码参照官网的rsocket-java/examples
demo的git地址：https://github.com/KouLouYiMaSi/rsocketdemo
Before
- 创建普通maven项目
- 引入必要依赖如下
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-transport-netty</artifactId>
<version>1.0.0</version>
</dependency>
Request-Response应用
- 服务端
import java.util.concurrent.TimeUnit;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Mono;
public class RequestResponseServer {
public static void main(String[] args) throws InterruptedException {
// RSocket类能够提供多种通讯模型，此处用到的就是request-response
RSocket rsocket = new RSocket() {
boolean fail = false;
// Mono表示 最多返回一个Payload，Payload裹挟着数据
@Override
public Mono<Payload> requestResponse(Payload p) {
if (fail) {
fail = false;
return Mono.error(new Throwable("发生异常"));
} else {
return Mono.just(DefaultPayload.create("这是服务端对：" + p.getDataUtf8() + "的返回"));
}
}
};
// RSocketServer是用于创建RSocket服务的类
// SocketAcceptor.with指定以rsocket的形式创建两端连接
// subscribe方法用于订阅连接上返回的消息
RSocketServer.create(SocketAcceptor.with(rsocket))
.bind(TcpServerTransport.create("localhost", 7000))
.subscribe();
// 由于server监听是用的其他线程，相当于异步，所以让主线程不要死掉
TimeUnit.MINUTES.sleep(10);
}
}
- 客户端
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
public class RequestResponseClient {
public static void main(String[] args) {
RSocket socketClient =
RSocketConnector.connectWith(TcpClientTransport.create("localhost", 7000)).block();
// Payload create用于构建消息
// doOnNext用于处理收到的连接上的信息
// block阻塞等待直到连接上有消息到来
for (int i = 0; i < 10; i++) {
socketClient
.requestResponse(DefaultPayload.create("客户端消息 " + i))
.map(Payload::getDataUtf8)
.onErrorReturn("error")
.doOnNext(s -> System.out.println(s))
.block();
}
// 切断连接
socketClient.dispose();
}
}
Fire-And-Forget应用
- 服务端代码
import java.util.concurrent.TimeUnit;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.publisher.Mono;
public class FireAndForgetServer {
public static void main(String[] args) throws InterruptedException {
RSocketServer.create(SocketAcceptor.forFireAndForget(payload -> {
System.out.println(payload.getDataUtf8());
return Mono.empty();
})).bind(TcpServerTransport.create("localhost", 7001))
.doOnSuccess(msg -> System.out.println(Thread.currentThread().getName()))
.doOnNext(msg -> System.out.println(msg))
.doOnError(msg -> System.out.println(msg))
.block();
TimeUnit.MINUTES.sleep(10);
}
}
- 客户端代码:注意要间隔1s发，直接发出去不能稳定的打出来，具体原因不确定是丢包还是缓存
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
public class FireAndForgetClient {
public static void main(String[] args) throws InterruptedException {
RSocket socketClient =
RSocketConnector.connectWith(TcpClientTransport.create("localhost", 7001)).block();
for (int i = 1; i < 11; i++) {
TimeUnit.SECONDS.sleep(1);
socketClient.fireAndForget(DefaultPayload.create("客户端消息FireAndForget"+i)).block();
}
}
}
Request-Stream应用
- 创建服务端
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;
public class RequestStreamServer {
public static void main(String[] args) throws InterruptedException {
// 收到请求之后返回消息流，Flux可以理解成多个Mono即多条消息
RSocketServer.create(
SocketAcceptor.forRequestStream(payload -> Flux.interval(Duration.ofMillis(100))
.map(aLong -> DefaultPayload.create("服务端返回消息: " + aLong))))
.bind(TcpServerTransport.create("localhost", 7000))
.subscribe();
TimeUnit.MINUTES.sleep(10);
}
}
- 创建客户端
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
public class RequestStreamClient {
public static void main(String[] args) {
RSocket socket =
RSocketConnector.connectWith(TcpClientTransport.create("localhost", 7000)).block();
// take（10）只取10条流数据
socket
.requestStream(DefaultPayload.create("Hello"))
.map(Payload::getDataUtf8)
.doOnNext(msg -> System.out.println(msg))
.take(10)
.then()
.doFinally(signalType -> socket.dispose())
.then()
.block();
}
}
Channel应用
- 服务端
import java.util.concurrent.TimeUnit;
import io.rsocket.Payload;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;
public final class ChannelServer {
public static void main(String[] args) throws InterruptedException {
SocketAcceptor serverAcceptor =
SocketAcceptor.forRequestChannel(
payloads -> Flux.from(payloads)
.map(Payload::getDataUtf8)
.map(s -> "服务端收到消息为: " + s)
.map(DefaultPayload::create));
RSocketServer.create(serverAcceptor)
.bind(TcpServerTransport.create("localhost", 7004))
.subscribe();
TimeUnit.MINUTES.sleep(10);
}
}
- 客户端
import java.time.Duration;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;
public class ChannelClient {
public static void main(String[] args) {
RSocket socket =
RSocketConnector.connectWith(TcpClientTransport.create("localhost", 7004)).block();
socket.requestChannel(
Flux.interval(Duration.ofMillis(1000)).map(i -> DefaultPayload.create("我是客户端啊")))
.map(Payload::getDataUtf8)
.doOnNext(msg -> System.out.println(msg))
.take(10)
.doFinally(signalType -> socket.dispose())
.then()
.block();
}
}
最后
- 不要被我类的名字迷惑，在RSocket中客户端与服务端是等同的，两个概念不要过度区分
- 本质上的使用就是创建两端连接，创建对连接上消息的响应，创建交互模型
