Spring 6 中的 RSocket 接口

教程

1、概览

本文将带你了解如何在 Spring 6 中使用 RSocket。

随着 Spring 6 引入声明式 RSocket 客户端，使用 RSocket 变得更加简单。这一功能消除了重复的模板代码，使开发人员能够更高效地使用 RSocket。

2、Maven 依赖

首先，创建 Spring Boot 项目，并在 pom.xml 文件中添加 spring-boot-starter-rsocket 依赖：

<dependency> 
    <groupId>org.springframework.boot</groupId> 
    <artifactId>spring-boot-starter-rsocket</artifactId>
    <version>3.1.4</version>
</dependency>

3、创建 RSocket Server

首先，创建一个 responder（应答器），通过 Controller 来管理传入的请求：

@MessageMapping("MyDestination")
public Mono<String> message(Mono<String> input) {
    return input.doOnNext(msg -> System.out.println("Request is:" + msg + ",Request!"))
      .map(msg -> msg + ",Response!");
}

接着，在 application.properties 文件中添加以下属性，以在 7000 端口启动服务（ MyDestination）：

spring.rsocket.server.port=7000

4、客户端

现在，开发客户端。为了简单起见，我们在同一个项目中创建客户端，但将其放在一个单独的包中。实际开发中，它们应该放在一个单独的项目中。

创建客户端接口：

public interface MessageClient {

    @RSocketExchange("MyDestination")
    Mono<String> sendMessage(Mono<String> input);
}

在使用客户端接口时，通过 @RSocketExchange 来指定 RSocket 端点。基本上，这意味着我们需要一些信息来建立端点路径。可以在接口级别上通过分配一个共享路径来实现。这非常简单，帮助我们知道要使用哪个端点。

5、测试

每个 Spring Boot 项目都包含一个使用 @SpringBootApplication 注解的类。当项目加载时，该类会运行。因此，我们可以使用这个类并添加一些 Bean 测试场景。

5.1、创建 RSocketServiceProxyFactory Bean

首先，需要创建一个 Bean 来生成 RSocketServiceProxyFactory。

该工厂负责创建 RSocket 服务接口的代理实例。它负责处理这些代理的创建，并通过指定服务器接收传入连接的主机和端口，建立与 RSocket 服务器的必要连接：

@Bean
public RSocketServiceProxyFactory getRSocketServiceProxyFactory(RSocketRequester.Builder requestBuilder) {
    RSocketRequester requester = requestBuilder.tcp("localhost", 7000);
    return RSocketServiceProxyFactory.builder(requester).build();
}

5.2、创建 MessageClient

然后，创建一个 Bean，负责生成客户端接口：

@Bean
public MessageClient getClient(RSocketServiceProxyFactory factory) {
    return factory.createClient(MessageClient.class);
}

5.3、创建 Runner Bean

最后，创建一个 Runner Bean，使用 MessageClient 实例发送和从服务器接收消息：

@Bean
public ApplicationRunner runRequestResponseModel(MessageClient client) {
    return args -> {
        client.sendMessage(Mono.just("Request-Response test "))
          .doOnNext(message -> {
              System.out.println("Response is :" + message);
          })
          .subscribe();
    };
}

5.4、测试结果

当通过命令行运行 Spring Boot 项目时，会显示以下结果：

>>c.b.r.responder.RSocketApplication : Started 
>>RSocketApplication in 1.127 seconds (process running for 1.398)
>>Request is:Request-Response test ,Request!
>>Response is :Request-Response test ,Response!

6、RSocket 交互模型

RSocket 是一种二进制协议，用于创建快速响应的分布式应用程序。它为服务器和客户端之间的数据交换提供了不同的通信模式。

通过这些交互模型，开发人员可以设计满足 Data Flow、Backlog 和特定需求的系统。

RSocket 有四种主要的交互模型可用。这些方法之间的主要区别基于输入和输出的数量。

6.1、请求-响应

在这种方法中，每个请求都会收到一个单独的响应。因此，我们使用了一个基数为 1 的 Mono 请求，并收到了一个具有相同基数的 Mono 响应。

到目前为止，本文中的所有代码都是基于此模型的。

6.2、请求-流

当我们订阅时事通讯时，我们会定期收到来自服务器的更新信息流。当客户端发出初始请求时，服务器会发送一个数据流作为响应。

请求可以是 Mono 或 Void，但响应总是 Flux：

@MessageMapping("Counter")
public Flux<String> Counter() {
    return Flux.range(1, 10)
      .map(i -> "Count is: " + i);
}

6.3、请求后不管

类似于信件投递，把信投进邮箱，并不指望收到回复。类似地，在这种模式（fire-and-forget）中，响应可以是 null 值或单个 Mono：

@MessageMapping("Warning")
public Mono<Void> Warning(Mono<String> error) {
    error.doOnNext(e -> System.out.println("warning is :" + e))
      .subscribe();
    return Mono.empty();
}

6.4、Channel

类似于对讲机，双方可以同时说话和监听。这种通信依赖于发送和接收 Data Flux：

@MessageMapping("channel")
public Flux<String> channel(Flux<String> input) {
    return input.doOnNext(i -> {
          System.out.println("Received message is : " + i);
      })
      .map(m -> m.toUpperCase())
      .doOnNext(r -> {
          System.out.println("RESPONSE IS :" + r);
      });
}

7、总结

在本文中，我们学习了 Spring 6 中新的声明式 RSocket 客户端功能，以及如何使用 @RSocketExchange 注解。

此外，我们还详细了解了如何创建和设置服务代理，以便使用 TCP 协议轻松安全地连接到远程端点。

参考：https://www.baeldung.com/spring-rsocket