netty-rpc 项目功能详解

0x01 优雅停机

优雅停机需要满足以下两个要求：

• 关闭服务提供者：不再接受服务请求，而对于正在执行的服务请求，等待其执行完毕之后再下线
• 关闭服务消费者：不再发送服务请求，而对于已经发送了的请求，需要等待其响应

netty-rpc 优雅停机的总流程如下所示：

netty-rpc 中实现的优雅停机机制主要包含 6 个步骤：

1. 收到kill -9进程退出信号，会触发注册在 JVM 中的 ShutdownHook 来执行优雅停机操作
2. 注销注册中心：会取消注册在 ZK 端的服务元数据信息，并且取消注册在 ZK 端的监听器，最后关闭掉和 ZK 端的连接。因此 ZK 端会通知客户端，服务已经下线，这就实现了服务端不再接受新的服务请求
3. 注销 Protocol：
   1. Provider 端会关闭掉 HeaderExchangeServer 中的心跳检测，并且会发送 readonly 事件报文通知 consumer 当前服务不可用，同样实现了服务端不再接受新的服务请求。
   2. Provider 端会关闭掉 NettyServer，不过会等待 NettyServer 中的线程池中正在执行的任务执行完毕，这就实现了对于正在执行的服务请求，等待其执行完毕之后再下线。
   3. Provider 端会关闭掉 jmx 服务器以及 http 服务器
   4. Consumer 端会首先将 ReferenceCountExchangeClient 中的引用计数减一，然后关闭掉 HeaderExchangeClient 中的心跳检测，并且在 HeaderExchangeClient 中，会检查 DefaultFuture 中是否还有当前 channel，如果有的话，说明还有请求在等待响应，就等待一段时间，这就实现了对于已经发送了的请求，等待其响应。接着关闭到服务端的连接，不会再发送新的请求到服务端。

0x02 实时监控服务调用情况

有两个关键字 <nettyrpc:service/> 中的 monitor 以及 <nettyrpc:application/> 中的 metrics，其中 metrics 用来表示是否开启全局的监控，默认是开启状态。而 monitor 用来表示是否开启对某一个方法的监控，默认也为 true，也就是开启状态。

服务监控是以类中的方法为单位的，会显示这个方法的调用次数、调用成功的次数、方法被调用的平均耗时、最大耗时以及最小耗时、调用失败的次数、方法调用失败的堆栈信息、上一次调用失败的时间。

对服务调用状态的监控是通过服务端拦截器 MonitorFilter 完成的。当对服务端的方法进行调用时，就会调用这个 MonitorFilter。在这个 MonitorFilter 中，监控的最小单位是类中的方法，并且类中方法的调用数据保存在一个 MetricsVisitor 对象中，这个对象内部保存了各个监控数据。在 MonitorFilter 中首先会让方法继续调用，获取到方法调用的结果，然后就根据类名和方法签名获取到对应的 MetricsVisitor 对象。接下来步骤是：

• 更新方法的调用次数
• 根据结果中是否有异常，来判断方法调用成功与否
  • 如果调用成功，那么就更新方法调用成功的次数、更新方法调用最大耗时、最小耗时以及平均耗时；
  • 如果调用失败就更新调用失败的次数、方法调用失败的堆栈信息、上一次调用失败的时间。

在获取 MetricsVisitor 对象时，涉及到并发控制的问题，因为可能会有多个线程同时想要获取到 Visitor 对象。具体的逻辑是使用了一个队列 waiters 来使得多个线程有序地，一次一个地进入到临界区域中，获取 visitor。

• enter 方法（进入临界区之前执行）具体的逻辑是只有 waiters 队列中的队头元素可以进入临界区，其余的线程只能阻塞等待，即首先将当前线程加入到队列中，然后循环判断当前线程是否是队列的头元素，如果不是的话，就阻塞等待。如果是的话，就执行完方法，进入临界区。
• exit 方法（退出临界区之后执行）具体逻辑就是将当前线程移出 waiters 队列头元素，并且 waiters 队列不为空的话，唤醒队列的头元素。

另外，在更新 MetricsVisitor 中的各个字段时，也会涉及到并发控制的问题，这里是通过 CAS 的方式来解决的。

0x03 注册中心缓存机制

调用者需要通过注册中心（例如：ZK）上的注册信息，获取提供者，但是如果频繁往 ZK 获取信息，肯定会存在单点故障问题，所以提供了将提供者信息缓存在本地的方法。当消费者无法往注册中心上注册监听器时（也就无法获取提供者的任何信息），就会通过缓存获取提供者的 url 信息，然后更新自己保存的 invoker 集合，消费者通过这些 invoker 集合发起 rpc 调用。

在订阅注册中心的回调处理逻辑当中，也就是当第一次把监听器注册到注册中心对应的目录上时或者之后目录下的结点发生了变更，那么就会返回这个目录下的全量数据。接着会把这个消费者订阅的某一个方法下所有 provider 返回的 url 全都合并起来（一个 RegistryDirectory 只能注册监听一个方法服务），并且以空格作为分隔符，然后保存这些服务提供者信息先保存到内存中的 properties 对象中，key 为 ServiceName，也就是服务的接口名。最后保存到本地缓存文件当中（同步/异步两种方式）。

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/>

• key:参数的键
• value:参数的值

通过 <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 服务器，可以让我们通过浏览器来访问服务的导出情况，以及可以屏蔽或者恢复服务。

具体的访问地址如下：

• http://169.254.207.250:10220/netty-rpc/ability
• http://169.254.207.250:10220/netty-rpc/metrics

第一个是服务能力展示的页面地址：

第二个是服务调用情况的页面地址：

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，具体如下：