首页 站点介绍

RSocket By Example

RSocket Java SDK

在RSocket众多开发语言SDK中，目前支持的最好就是Java SDK，所以这里我们介绍一下RSocket Java SDK。 RSocket Java SDK是基于Reactor这款Java Reactive框架开发，所以Reactor的一些编辑特性在SDK中都有体现。 另外RSocket得到Spring社区的大力支持，如Spring Framework 5.2+内置RSocket支持，Spring Boot 2.2+的RSocket Starter将RSocket开发变得更加简单。 相信不久你可以看多更多的产品会支持RSocket协议。

如果你想马上尝试一下RSocket，可以参考 Getting Started With RSocket: Spring Boot Server

当然最快捷的方式尝试RSocket Java就是使用JBang，样例项目地址为： https://github.com/linux-china/jbang-rsocket

///usr/bin/env jbang "$0" "$@" ; exit $?
//JAVA 8
//DEPS org.slf4j:slf4j-simple:1.7.36
//DEPS io.rsocket:rsocket-core:1.1.2
//DEPS io.rsocket:rsocket-transport-netty:1.1.2

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.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;

public class ServerExample {
    public static void main(String[] args) {
        Hooks.onErrorDropped(e -> {});
        RSocket handler = new RSocket() {
            @Override
            public Mono<Payload> requestResponse(Payload payload) {
                System.out.println("RequestResponse: " + payload.getDataUtf8());
                return Mono.just(payload);
            }

            @Override
            public Flux<Payload> requestStream(Payload payload) {
                System.out.println("RequestStream: " + payload.getDataUtf8());
                return Flux.just("First", "Second").map(DefaultPayload::create);
            }

            @Override
            public Mono<Void> fireAndForget(Payload payload) {
                System.out.println("FireAndForget: " + payload.getDataUtf8());
                return Mono.empty();
            }
        };
        RSocketServer.create(SocketAcceptor.with(handler))
                .bindNow(TcpServerTransport.create("localhost", 7000))
                .onClose()
                .doOnSubscribe(subscription -> System.out.println("RSocket Server listen on tcp://localhost:7000"))
                .block();
    }
}

这里给出一些常见样例：