netty-rpc 项目解析
0x01 简介
使用 Netty、Zookeeper 和 Spring,实现了一个轻量级的 RPC 框架,RPC 使用 Netty 作为网络通讯框架,Zookeeper 作为注册中心进行服务的自动注册和发现,Spring 作为容器,经过简单测试,QPS 大概为 1K~2K。
0x02 功能
netty-rpc 项目实现的功能如下所示:
- zookeeper作为注册中心的服务注册和发现
- 内置3种负载均衡:加权轮询(WeightedRoundRobin)、加权随机(WeightedRandom)、一致性哈希(ConsistentHash)
- 支持多种序列化协议,比如jdk和protostuff
- 内置4种集群容错机制:失败自动切换(Failover)、失败自动恢复(Failback)、快速失败(Failfast)、失败安全(Failsafe)
- 心跳机制
- 超时机制
- 断线重连机制
- 服务能力展示
- 桩
- 服务降级
- 服务端和客户端过滤器
- 注册中心缓存
- url直连
- 结果缓存
- SPI机制
- 令牌验证
- 限流
- 粘滞连接
- TCP/InJVM协议
- 同步与异步调用
- JMX监控
- 优雅停机
0x03 标签和使用
在netty-rpc框架中,目前只支持XML文件进行配置。配置的标签和属性如下:
<nettyrpc:application/>
:metrics、name、metricsPort、echoPort
<nettyrpc:protocol/>
:id, name, port, serialize, host
<nettyrpc:registry/>
: id, address, name, file
<nettyrpc:service/>
: id, interfaceName, ref, filter, token, cache, timeout, monitor, weight, scope, protocol, limiter, rate
<nettyrpc:reference/>
: id, interfaceName, registry, tiemout, retries, loadbalance, async, cluster, stub, scope, filter, protocol, cache, sticky, url, mock
<nettyrpc:parameter/>
: key, value
下面分别介绍每一个标签,以及其中每一个属性的作用。
1 <nettyrpc:application/>
- metrics:是否开启监控,如果配置为false,则会关闭监控;如果不配置或者配置为true,那么就会开启监控
- name:服务者段的名字
- metricsPort:启动jmx服务器所需要的端口,不配置的话,默认使用9999
- echoPort:启动echo服务器所需要的端口,不配置的话,默认使用18882
2 <nettyrpc:protocol/>
- id: 配置的协议的id
- name:配置的协议的名称,比如rpc,injvm
- port:协议所使用的端口号
- serialize:协议所使用的序列化方式
- host:协议导出的地址,不配置的话,默认使用本机地址
3 <nettyrpc:registry/>
- id:注册中心的id
- name:注册中心的名称,比如zookeeper、nacos
- address:注册中心的地址,一般为协议名://ip地址:端口号
- file: 注册中心保存缓存信息的文件地址
4 <nettyrpc:service/>
- id:导出服务的 id
- interfaceName:服务接口名
- ref:服务对象实现引用
- filter:过滤器
- token:令牌验证,为空表示不开启,如果为 true,表示 UUID 随机生成动态令牌,否则使用静态令牌,令牌的作用是防止消费者绕过注册中心直接访问,保证注册中心的授权功能有效,如果使用点对点调用,需关闭令牌功能
- registry:向指定的注册中心进行注册,在多个注册中心时使用,值为
<nettyrpc:registry>
的 id 属性,多个注册中心 ID 用逗号分隔
- cache:使用到的缓存的种类,比如 lru、threadlocal,配置了的话,会激活使用对应种类的缓存过滤器,对方法的调用结果进行缓存
- timeout:远程服务调用超时时间
- monitor:表示此服务是否开启了监控,默认为 true,也就是开启监控,如果为 false 的话,则关闭掉对这个服务的监控。如果
<nettyrpc:application/>
中 metrics 为 false的话,这里的 monitor 配置就不会起效果,监控始终是被关闭的
- weight:此服务的权重,用于负载均衡
- scope:服务导出的范围,可以为 local 或者 remote,local 表明只导出到本地,而 remote 只导出到远程,如果不配置 scope 的话,那么默认会同时导出到远程和本地
- protocol:使用指定的协议暴露服务,在多协议时使用,值为
<nettyrpc:protocol>
的 id 属性,多个协议ID用逗号分隔
- limiter:限流器的种类,可以为 bucket、token 以及 guava
- rate:限流的速率
5 <nettyrpc:registry/>
- id:导出服务的 id
- interfaceName:服务接口名
- registry:向指定的注册中心进行注册,在多个注册中心时使用,值为
<nettyrpc:registry>
的 id 属性,多个注册中心 ID 用逗号分隔
- timeout:远程服务调用超时时间
- retries:表明重试的次数,不包含第一次调用
- loadbalance:表明使用的负载均衡策略,比如 hash、roundrobin、random,默认使用 random,也就是加权随机负载均衡
- async:表明是否使用异步调用
- cluster:表明使用的容错机制,比如 failover,failback,failfast,failsafe 这四种
- stub:桩,值可以为 true/false/类名,如果为 true 的话,会去同样包下寻找带有 Stub 后缀的类,如果为类名的话,直接使用这个类作为桩
- scope:服务引用的范围,可以为 local 或者 remote,local 表明只引用本地的服务,而 remote 则会引用远程服务,也就是注册中心上的服务,如果不配置 scope 的话,那么默认会引用远程服务
- filter:过滤器
- protocol:只调用指定协议的服务提供方,其它协议忽略
- cache:使用到的缓存的种类,比如 lru、threadlocal,配置了的话,会激活使用对应种类的缓存过滤器,对方法的调用结果进行缓存
- sticky:是否开启粘滞连接,粘滞连接用于有状态服务,尽可能让客户端总是向同一提供者发起调用,除非该提供者挂了,再连另一台
- url:直连服务提供者,直接指明服务端的地址以及协议,多个提供者使用分号分隔开,比如 rpc://169.224.214.250:10886;rpc://169.224.214.250:10887
- mock:服务降级,它可以配置的值为 true/false/force/fail/类名
6 <nettyrpc:parameter/>
通过 <nettyrpc:parameter/>
标签配置的参数值:
<nettyrpc:parameter key="heartbeat" value="4000"/> <!--也就是发送心跳包的时间间隔-->
<nettyrpc:parameter key="heartbeatTimeout" value="30000"/> <!--心跳超时的时间,也就是如果在这个时间内没有接收到对方发送过来的心跳包或者其它数据包,客户端会进行重连,服务器则会直接断开连接-->
<nettyrpc:parameter key="capacity" value="3"/> <!--缓存的大小,用于lru缓存的参数配置-->
<nettyrpc:parameter key="segments" value="2"/> <!--缓存的段的个数,也是用于lru缓存的参数配置,lru缓存在实现时,分为多个段,每一个段都用读写锁控制并发,不同的段之间可以进行并发操作。上面的capacity主要用于控制每一个段的缓存大小-->
0x04 调用流程-客户端的图
用户使用框架生成的对象调用对应的方法,框架生成的这个对象是一个使用 JDK 动态代理机制生成的对象,它实现了用户接口,调用接口里面的方法会被转移到调用 ProxyWrapper 里面的 invoke 方法。MockClusterInvoker 中包含了服务降级的逻辑,接着又是FailsafeClusterInvoker,这个包含了集群容错逻辑,它会使用负载均衡从多个 Invoker 中选出一个,然后调用选出的 RpcInvoker 的 invoke 方法。
在 RpcInvoker 中调用 ReferenceCountClient 的 request 方法,这里的 ReferenceCounClient 只是对 NettyClient 做了一个封装,增加了计数逻辑。这是因为,使用共享连接,也就是访问同一个地址的服务端,多个 RpcInvoker 会使用同一个 NettyClient,因此当其中一个关闭时,如果还有其它 RpcInvoker 在访问,那么就不能将其真正关闭。
接下来是 HeaderExchangeClient 的 request 方法,HeaderExchangeClient 中有心跳逻辑。接下来就是通过 NettyClient 真正的发送请求到服务端。NettyClient 的 pipeline 结构以及各个 handler 稍后会讲到。
0x05 调用流程-服务端的图
客户端发送过来的请求被解码之后,传递给 NettyServerHandler 的 channelRead 方法,在 NettyServerHandler 中又封装了 HeartbeatHandler 和 ExchangeHandler。在 ExchangeHandler中,它会接收客户端发过来的 Rpc 调用请求,并且将方法的实际调用包装成一个 Task,也就是 RecvExecutionTask,然后放到线程池中去执行。不会阻塞 Netty 的 worker 线程。
线程池中,RecvExecutionTask 的执行流程如下:
RecvExecutionTask 在被线程池中执行时,会调用 RpcProtocol 的一个匿名内部类,它实现了 ReplyHandler 接口,调用其 reply 方法。接着会经过一系列的过滤器,也就是 ChainFilter,比如 MonitorChainFilter、CacheChainFilter。然后通过反射调用执行对应的方法。
0x06 服务导出流程
通过 contextRefreshedEvent 触发服务导出,每一个 NettyRpcService 对应一个 <nettyrpc:service/>
标签。在导出时,会把每一个服务使用指定的所有协议导出到所有的注册中心上。在导出时,默认会将服务导出到远程和本地,如果指定 scope 属性为remote 或者 local 的话,就只会将服务导出到远程或者本地。在导出到远程时,会将服务注册到注册中心上,同时开启服务器,也就是开启 NettyServer,进行监听,同时也会开启 JmxServer,它主要是让用户可以通过 Jconsole 来查看具体调用情况,同时也可以通过浏览器来访问服务的具体调用信息。与此同时,会开启一个内置的 HTTP 服务器,可以让我们通过浏览器来访问服务的导出情况,以及可以屏蔽或者恢复服务。
具体的访问地址如下:
第一个是服务能力展示的页面地址:
第二个是服务调用情况的页面地址:
0x07 服务引用流程
NettyRpcReference 类实现了 FactoryBean 接口,因此只有在调用 ctx.getBean 的时候,才会去调用 NettyRpcReference 的 geObject 方法。然后在 ReferenceConfig 的 init 方法中,主要是获取一些配置参数,接着在 createProxy 方法中,分为三种情况,一是本地引用还是远程引用,对于远程引用又分为两种,分别是 url 直连和从注册中心上获取 url。
如果是从注册中心上获取 url 的话,首先得将一个监听器注册在注册中心的某个目录上,比如 /rpc/com.xxx.DemoService,然后目录下面就直接是提供者的 url。而这些提供者的 url 会使用一个服务目录,RegistryDirectory 保存起来。另外,如果有多个注册中心或者多个服务直连 url,那么将会使用到 AvailableCluster 将其聚合起来。
0x08 包的结构
netty-rpc-core 里面的是框架的源代码,java 目录的结构如下:
- async: 异步代码,比如 DefaultRpcFuture
- cluster: 集群代码,比如负载均衡,集群容错以及各种服务目录
- commons: 工具类代码,包括 jmx,cache,ratelimiter 等
- config: 配置类和 Spring 标签解析类代码
- core: 包含了
@Extension、@Attribute、@Activate
注释,以及 ExtensionLoader 代码等
- filter: 包含了过滤器的代码
- protocol: 包含了 InjvmProtocol、RpcProtocol、RegistryProtocol 协议代码,以及对应的 Invoker 和 Exporter 代码
- registry: 包含了 Zookeeper 注册中心的代码
- remoting: 包含了 Client、Server 以及 Handler 的代码
- serialize:包含了各种序列化的代码,包括 jdk、protostuff
resources 包的结构如下:
0x09 网络结构
客户端的 pipeline 结构为:MessageEncoder -> MessageDecoder -> NettyClientHandler。而 NettyClientHandler 又封装了多个 ChannelHandler,具体如下:
其中 HeartbeatHandler 会对心跳的请求和响应进行处理,其它请求则继续传递给后面的 ChannelHandler 来处理,也就是ExchangeHandler,它会把客户端发过来的Rpc调用请求,并且将方法的实际调用包装成一个 Task 放入到线程池进行处理,如果是响应的话,则调用 DefaultRpcFuture 的 received 方法,表示接收到服务端的响应。
服务端的 pipeline 结构为:MessageEncoder -> MessageDecoder -> NettyServerHandler
而 NettyServerHandler 也封装了多个 ChannelHandler,具体如下: