如何使用Java开发一个基于RSocket的异步通信应用

<dependency>
    <groupId>io.rsocket</groupId>
    <artifactId>rsocket-core</artifactId>
    <version>1.1.0</version>
</dependency>

import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class RSocketServer {

    public static void main(String[] args) {
        CloseableChannel closeableChannel = RSocketFactory.receive()
                .acceptor((setup, sendingSocket) -> Mono.just(new RSocketHandler()))
                .transport(TcpServerTransport.create("localhost", 8080))
                .start()
                .block();

        // Prevent the application from terminating
        closeableChannel.onClose().block();
    }

    static class RSocketHandler extends AbstractRSocket {

        @Override
        public Mono<Void> fireAndForget(Payload payload) {
            System.out.println("Received fire-and-forget request: " + payload.getDataUtf8());
            // Process the request and return void
            return Mono.empty();
        }

        @Override
        public Mono<Payload> requestResponse(Payload payload) {
            System.out.println("Received request-response request: " + payload.getDataUtf8());
            // Process the request and return a response
            String response = "Hello, " + payload.getDataUtf8();
            return Mono.just(DefaultPayload.create(response));
        }

        @Override
        public Flux<Payload> requestStream(Payload payload) {
            System.out.println("Received request-stream request: " + payload.getDataUtf8());
            // Process the request and return a stream of responses
            String response = "Hello, " + payload.getDataUtf8();
            return Flux.just(DefaultPayload.create(response));
        }
    }
}

import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class RSocketClient {

    public static void main(String[] args) {
        RSocket rSocket = RSocketFactory.connect()
                .transport(TcpClientTransport.create("localhost", 8080))
                .start()
                .block();

        // Send a fire-and-forget request
        rSocket.fireAndForget(DefaultPayload.create("World")).block();

        // Send a request-response request
        Mono<Payload> responseMono = rSocket.requestResponse(DefaultPayload.create("World"));
        responseMono.subscribe(response -> System.out.println("Received response: " + response.getDataUtf8()));

        // Send a request-stream request
        Flux<Payload> responseFlux = rSocket.requestStream(DefaultPayload.create("World"));
        responseFlux.subscribe(response -> System.out.println("Received response: " + response.getDataUtf8()));
    }
}