RSocket協議初識-Java中使用（二）

前言

在RSocket協議初識（一）中我們使用SpringBoot創(chuàng)建了簡單的RSocket通訊模型，本篇我們將介紹純Java使用Demo。本文代碼參照官網的rsocket-java/examples
demo的git地址：https://github.com/KouLouYiMaSi/rsocketdemo

Before

  • 創(chuàng)建普通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("發(fā)生異常"));
                } else {
                    return Mono.just(DefaultPayload.create("這是服務端對：" + p.getDataUtf8() + "的返回"));
                }
            }
        };

        // RSocketServer是用于創(chuàng)建RSocket服務的類
        // SocketAcceptor.with指定以rsocket的形式創(chuàng)建兩端連接
        // subscribe方法用于訂閱連接上返回的消息
        RSocketServer.create(SocketAcceptor.with(rsocket))
                .bind(TcpServerTransport.create("localhost", 7000))
                .subscribe();
        // 由于server監(jiān)聽是用的其他線程，相當于異步，所以讓主線程不要死掉
        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發(fā)，直接發(fā)出去不能穩(wěn)定的打出來，具體原因不確定是丟包還是緩存
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應用

  • 創(chuàng)建服務端
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);
    }
}
  • 創(chuàng)建客戶端
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中客戶端與服務端是等同的，兩個概念不要過度區(qū)分
  • 本質上的使用就是創(chuàng)建兩端連接，創(chuàng)建對連接上消息的響應，創(chuàng)建交互模型
最后編輯于 ：
?著作權歸作者所有,轉載或內容合作請聯系作者
【社區(qū)內容提示】社區(qū)部分內容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內容（如有圖片或視頻亦包括在內）由作者上傳并發(fā)布，文章內容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

友情鏈接更多精彩內容