写点什么

Spring RSocket:基于服务注册发现的 RSocket 负载均衡

发布于: 2021 年 02 月 19 日


作者 | 雷卷

来源|阿里巴巴云原生公众号


RSocket 分布式通讯协议是 Spring Reactive 的核心内容,从 Spring Framework 5.2 开始,RSocket 已经是 Spring 的内置功能,Spring Boot 2.3 也添加了 spring-boot-starter-rsocket,简化了 RSocket 的服务编写和服务调用。RSocket 通讯的核心架构中包含两种模式,分别是 Broker 代理模式和服务直连通讯模式。


Broker 的通讯模式更灵活,如 Alibaba RSocket Broker,采用的是事件驱动模型架构。而目前更多的架构则是面向服务化设计,也就是我们常说的服务注册发现和服务直连通讯的模式,其中最知名的就是 Spring Cloud 技术栈,涉及到配置推送、服务注册发现、服务网关、断流保护等等。在面向服务化的分布式网络通讯中,如 REST API、gRPC 和 Alibaba Dubbo 等,都与 Spring Cloud 有很好地集成,用户基本不用关心服务注册发现和客户端负载均衡这些底层细节,就可以完成非常稳定的分布式网络通讯架构。


RSocket 作为通讯协议的后起之秀,核心是二进制异步化消息通讯,是否也能和 Spring Cloud 技术栈结合,实现服务注册发现、客户端负载均衡,从而更高效地实现面向服务的架构?这篇文章我们就讨论一下 Spring Cloud 和 RSocket 结合实现服务注册发现和负载均衡。


服务注册发现


服务注册发现的原理非常简单,主要涉及三种角色:服务提供方、服务消费者和服务注册中心。典型的架构如下:



服务提供方,如 RSocket Server,在应用启动后,会向服务注册中心注册应用相关的信息,如应用名称,ip 地址,Web Server 监听端口号等,当然还会包括一些元信息,如服务的分组(group),服务的版本号(version),RSocket 的监听端口号,如果是 WebSocket 通讯,还需要提供 ws 映射路径等,不少开发者会将服务提供方的服务接口列表作为 tags 提交给服务注册中心,方便后续的服务查询和治理。


在本文中,我们采用 Consul 作为服务注册中心,主要是 Consul 比较简单,下载后执行 consul agent -dev 就可以启动对应的服务,当然你可以使用 Docker Compose,配置也非常简单,然后 docker-compose up -d 就可以启动 Consul 服务。


当我们向服务中心注册和查询服务时,都需要有一个应用名称,对应到 Spring Cloud 中,也就是 Spring Boot 对应的 spring.application.name 的值,这里我们称之为应用名称,也就是后续的服务查找都是基于该应用名称进行的。如果你调用 ReactiveDiscoveryClient.getInstances(String serviceId); 查找服务实例列表时,这个 serviceId 参数其实就是 Spring Boot 的应用名称。考虑到服务注册和后续的 RSocket 服务路由的配合以及方便大家理解,这里我们打算设计一个简单的命名规范。


假设你有一个服务应用,功能名称为 calculator,同时提供两个服务: 数学计算器服务(MathCalculatorService)和汇率计算器服务(ExchangeCalculatorService),  那么我们该如何来命名该应用及其对应的服务接口名?


这里我们采用类似 Java package 命名规范,采用域名倒排的方式,如 calculator 应用对应的则为 com-example-calculator 样式,为何是中划线,而不是点?. 在 DNS 解析中作为主机名是非法的,只能作为子域名存在,不能作为主机名,而目前的服务注册中心设计都遵循 DNS 规约,所以我们采用中划线的方式来命名应用。这样采用域名倒排和应用名结合的方式,可以确保应用之间不会重名,另外也方便和 Java Package 名称进行转换,也就是 -  和 . 之间的相互转换。


那么应用包含的服务接口应该如何命名?服务接口全名是由应用名称和 interface 名称组合而成,规则如下:


String serviceFullName = appName.replace("-", ".") + "." + serviceInterfaceName; 


例如以下的服务命名都是合乎规范的:


  • com.example.calculator.MathCalculatorService


  • com.example.calculator.ExchangeCalculatorService


**而 com.example.calculator.math.MathCalculatorService 则是错误的,  因为在应用名称和接口名称之间多了 math**。为何要采用这种命名规范?首先让我们看一下服务消费方是如何调用远程服务的。假设服务消费方拿到一个服务接口,如 com.example.calculator.MathCalculatorService,那么他该如何发起服务调用呢?


  • 首先根据 Service 全面提取处对应的应用名称(appName),如 com.example.calculator.MathCalculatorService 服务对应的 appName 则为 com-example-calculator。如果应用和服务接口之间不存在任何关系,那么想要获取服务接口对应的服务提供方信息,你可能还需要应用名称,这会相对来说比较麻烦。如果接口名称中包含对应的应用信息,则会简单很多,你可以理解为应用是服务全面中的一部分。


  • 调用 ReactiveDiscoveryClient.getInstances(appName) 获取应用名对应的服务实例列表(ServiceInstance),ServiceInstance 对象会包含诸如 IP 地址,Web 端口号、RSocket 监听端口号等其他元信息。


  • 根据 RSocketRequester.Builder.transports(servers) 构建具有负载均衡能力的 RSocketRequester 对象。


  • 使用服务全称和具体功能名称作为路由进行 RSocketRequester 的 API 调用,样例代码如下:


rsocketRequester .route("com.example.calculator.MathCalculatorService.square")  .data(number)  .retrieveMono(Integer.class)


通过上述的命名规范,我们可以从服务接口全称中提取出应用名,然后和服务注册中心交互查找对应的实例列表,然后建立和服务提供者的连接,最后基于服务名称进行服务调用。该命名规范,基本做到到了最小化的依赖,开发者完全是基于服务接口调用,非常简单。


RSocket 服务编写


有了服务的命名规范和服务注册,编写 RSocket 服务,这个还是非常简单,和编写一个 Spring Bean 没有任何区别。引入 spring-boot-starter-rsocket 依赖,创建一个 Controller 类,添加对应的 MessagMapping annotation 作为基础路由,然后实现功能接口添加功能名称,样例代码如下:


@Controller @MessageMapping("com.example.calculator.MathCalculatorService") public class MathCalculatorController implements MathCalculatorService {     @MessageMapping("square")     public Mono<Integer> square(Integer input) {         System.out.println("received: " + input);         return Mono.just(input * input);     } }


上述代码看起来好像有点奇怪,既然是服务实现,添加 @Controller 和 @MessageMapping,看起来好像有点不伦不类的。当然这些 annotation 都是一些技术细节体现,你也能看出,RSocket 的服务实现是基于 Spring Message 的,是面向消息化的。这里我们其实只需要添加一个自定义的 @SpringRSocketService annotation 就可以解决这个问题,代码如下:


@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Controller @MessageMapping() public @interface SpringRSocketService {     @AliasFor(annotation = MessageMapping.class)     String[] value() default {}; }


回到服务对应的实现代码,我们改为使用 @SpringRSocketService annotation,这样我们的代码就和标准的 RPC 服务接口完全一模一样啦,也便于理解。此外 @SpringRSocketService 和 @RSocketHandler 这两个 Annotation,也方便我们后续做一些 Bean 扫描、IDE 插件辅助等。


@SpringRSocketService("com.example.calculator.MathCalculatorService") public class MathCalculatorImpl implements MathCalculatorService {     @RSocketHandler("square")     public Mono<Integer> square(Integer input) {         System.out.println("received: " + input);         return Mono.just(input * input);     } }


最后我们添加一下 spring-cloud-starter-consul-discovery 依赖,设置一下 bootstrap.properties,然后在 application.properties 设置一下 RSocket 监听的端口和元信息,我们还将该应用提供的服务接口列表作为 tags 传给服务注册中心,当然这个也是方便我们后续的服务管理。样例如下:


spring.application.name=com-example-calculator spring.cloud.consul.discovery.instance-id=com-example-calculator-${random.uuid} spring.cloud.consul.discovery.prefer-ip-address=true server.port=0 spring.rsocket.server.port=6565 spring.cloud.consul.discovery.metadata.rsocketPort=${spring.rsocket.server.port} spring.cloud.consul.discovery.tags=com.example.calculator.ExchangeCalculatorService,com.example.calculator.MathCalculatorService


RSocket 服务应用启动后,我们在 Consul 控制台就可以看到服务注册上来的信息,截屏如下:



RSocket 客户端接入


客户端接入稍微有一点复杂,主要是要基于服务接口全面要做一系列相关的操作,但是前面我们已经有了命名规范,所以问题也不大。客户端应用同样会接入服务注册中心,这样我们就可以获得 ReactiveDiscoveryClient bean,接下来就是根据服务接口全名,如 com.example.calculator.ExchangeCalculatorService 构建出具有负载均衡的 RSocketRequester。


原理也非常简单,前面说过,根据服务接口全称,获得其对应的应用名称,然后调用 ReactiveDiscoveryClient.getInstances(appName) 获得服务应用对应的实例列表,接下来将服务实例(ServiceInstance)列表转换为 RSockt 的 LoadbalanceTarget 列表,其实就是 POJO 转换,最后将转 LoadbalanceTarget 列表进行 Flux 封装(如使用 Sink 接口),传递给 RSocketRequester.Builder 就完成具有负载均衡能力的 RSocketRequester 构建,详细的代码细节大家可以参考项目的代码库。


这里要注意的是接下来如何感知服务端实例列表的变化,如应用上下线,服务暂停等。这里我采用一个定时任务方案,定时查询服务对应的地址列表。当然还有其他的机制,如果是标准的 Spring Cloud 服务发现接口,目前是需要客户端轮询的,当然也可以结合 Spring Cloud Bus 或者消息中间件,实现服务端列表变化的监听。如果客户端感知到服务列表的变化,只需要调用 Reactor 的 Sink 接口发送新的列表即可,RSocket Load Balance 在感知到变化后,会自动做出响应,如关闭即将失效的连接、创建新的连接等工作。


在实际的应用之间的相互通讯,会存在一些服务提供方不可用的情况,如服务方突然宕机或者其网络不可用,这就导致了服务应用列表中部分服务不可用,那么 RSocket 这个时候会如何处理?不用担心,RSocket Load Balance 有重试机制,当一个服务调用出现连接等异常,会重新从列表中获取一个连接进行通讯,而那个错误的连接也会标识为可用性为 0,不会再被后续请求所使用。服务列表推送和通讯期间的容错重试机制,这两者保证了分布式通讯的高可用性。


最后让我们启动 client-app,然后从客户端发起一个远程的 RSocket 调用,截屏如下:



上图中 com-example-calculator 服务应用包括三个实例,服务的调用会在这三个服务实例交替进行(RoundRobin 策略)。


开发体验的一些考量


虽然服务注册和发现、客户端的负载均衡这些都完成啦,调用和容错这些都没有问题,但是还有一些使用体验上的问题,这里我们也阐述一下,让开发体验做的更好。


1. 基于服务接口通讯


大多数 RPC 通讯都是基于接口的,如 Apache Dubbo、gRPC 等。那么 RSocket 能否做到?答案是其实完全可以。在服务端,我们已经是基于服务接口来实现 RSocket 服务啦,接下来我们只需要在客户端实现基于该接口的调用就可以。对于 Java 开发者来说,这不是大问题,我们只需要基于 Java Proxy 机制构建就可以,而 Proxy 对应的 InvocationHandler 会使用 RSocketRequester 来实现 invoke() 的函数调用。详细的细节请参考应用代码中的的 RSocketRemoteServiceBuilder.java 文件,而且在 client-app module 中也已经包含了解基于接口调用的 bean 实现。


2. 服务接口函数的单参数问题


使用 RSocketRequester 调用远程接口时,对应的处理函数只能接受单个参数,这个和 gRPC 的设计是类似的,当然也考虑了不同对象序列化框架的支持问题。但是考虑到实际的使用体验,可能会涉及到多参函数的情况,让调用方开发体验更好,那么这个时候该如何处理?其实从 Java 1.8 后,interface 是允许增加 default 函数的,我们可以添加一些体验更友好的 default 函数,而且还不影响服务通讯接口,样例如下:


public interface ExchangeCalculatorService {     double exchange(ExchangeRequest request);     default double rmbToDollar(double amount) {         return exchange(new ExchangeRequest(amount, "CNY", "USD"));     } }


通过 interface 的 default method,我们可以为调用方提供给便捷函数,如在网络传输的是字节数组 (byte[]),但是在 default 函数中,我们可以添加 File 对象支持,方便调用方使用。Interface 中的函数 API 负责服务通讯规约,default 函数来提升使用方的体验,这两者的配合,可以非常容易解决函数多参问题,当然 default 函数在一定程度上还可以作为数据验证的前哨来使用。


3. RSocket Broker 支持


前面我们说到,RSocket 还有一种 Broker 架构,也就是服务提供方是隐藏在 Broker 之后的,请求主要是由 Broker 承接,然后再转发给服务提供方处理,架构样例如下:



那么基于服务发现的机制负载均衡,能否和 RSocket Broker 模式混合使用呢?如一些长尾或者复杂网络下的应用,可以注册到 RSocket Broker,然后由 Broker 处理请求调用和转发。这个其实也不不复杂,前面我们说到应用和服务接口命名规范,这里我们只需要添加一个应用名前缀就可以解决。假设我们有一个 RSocker Broker 集群,暂且我们称之为 broker0 集群,当然该 broker 集群的实例也都注册到服务注册中心(如 Consul)啦。那么在调用 RSocket Broker 上的服务时,服务名称就被调整为 broker0:com.example.calculator.MathCalculatorService,也就是服务名前添加了 appName: 这样的前缀,这个其实是 URI 的另一种规范形式,我们就可以提取冒号之前的应用名,然后去服务注册中心查询获得应用对应的实例列表。 


回到 Broker 互通的场景,我们会向服务注册中心查询 broker0 对应的服务列表,然后和 broker0 集群的实例列表创建连接,这样后续基于该接口的服务调用就会发送给 Broker 进行处理,也就是完成了服务注册发现和 Broker 模式的混合使用的模式。


借助于这种定向指定服务接口和应用间的关联,也方便我们做一些 beta 测试,如你想将 com.example.calculator.MathCalculatorService 的调用导流到 beta 应用,你就可以使用 com-example-calculator-beta1:com.example.calculator.MathCalculatorService 这种方式调用服务,这样服务调用对应的流量就会转发给 com-example-calculator-beta1 对应的实例,起到 beta 测试的效果。


回到最前面说到的规范,如果应用名和服务接口的绑定关系你实在做不到,那么你可以使用这种方式实现服务调用,如 calculator-server:com.example.calculator.math.MathCalculatorService,只是你需要更完整的文档说明,当然这种方式也可以解决之前系统接入到目前的架构上,应用的迁移成本也比较小。如果你之前的面向服务化架构设计也是基于 interface 接口通讯的,那么通过该方式迁移到 RSocket 上完全没有问题,对客户端代码调整也最小。


总结


通过整合服务注册发现,结合一个实际的命名规范,就完成了服务注册发现和 RSocket 路由之间的优雅配合,当然负载均衡也是包含其中啦。对比其他的 RPC 方案,你不需要引入 RPC 自己的服务注册中心,复用 Spring Cloud 的服务注册中心就可以,如 Alibaba Nacos, Consul, Eureka 和 ZooKeeper 等,没有多余的开销和维护成本。如果你想更多了解 RSocket RPC 相关的细节,可以参考 Spring 官方博客 《Easy RPC with RSocket》


欢迎加入 alibaba-rsocket-broker 钉钉群:



更多详细的代码细节,可以点击链接查看文章对应的代码库!


发布于: 2021 年 02 月 19 日阅读数: 29
用户头像

阿里巴巴云原生 2019.05.21 加入

还未添加个人简介

评论

发布
暂无评论
Spring RSocket:基于服务注册发现的 RSocket 负载均衡