前言
SOFA-RPC 中对服务地址的选择也抽象为了一条处理链,由每一个 Router 进行处理。同 Filter 一样, SOFA-RPC 对 Router 提供了同样的扩展能力。
那么就看看 SOFA 是如何处理的。
如何使用
官方教程如下:
@Extension(value = "customerRouter")
@AutoActive(consumerSide = true)
public class CustomerRouter extends Router {
@Override
public void init(ConsumerBootstrap consumerBootstrap) {
}
@Override
public boolean needToLoad(ConsumerBootstrap consumerBootstrap) {
return ture;
}
@Override
public List<ProviderInfo> route(SofaRequest request, List<ProviderInfo> providerInfos) {
return providerInfos;
}
新建扩展文件 META-INF/services/sofa-rpc/com.alipay.sofa.rpc.client.Router 。内容如下:
customerRouter=com.alipay.sofa.rpc.custom.CustomRouter
如上自定义了一个 CustomerRouter ,生效于所有消费者。其中 init 参数 ConsumerBootstrap 是引用服务的包装类,能够拿到 ConsumerConfig ,代理类,服务地址池等对象。 needToLoad 表示是否生效该 Router , route 方法即筛选地址的方法。
可以看到,Router 也是通过 SOFA 的扩展机制实现的,通过定义一个 SPI 文件,能够有效的解耦。
然后我们再来看看他的原理。
源码解析
在 SOFA 中, Router 是个抽象类,内部定义了 4 个方法:
//初始化
public void init(ConsumerBootstrap consumerBootstrap) {
}
//是否自动加载
public boolean needToLoad(ConsumerBootstrap consumerBootstrap) {
return true;
}
// 筛选Provider
public abstract List<ProviderInfo> route(SofaRequest request, List<ProviderInfo> providerInfos);
//记录路由路径记录
protected void recordRouterWay(String routerName) {
if (RpcInternalContext.isAttachmentEnable()) {
RpcInternalContext context = RpcInternalContext.getContext();
String record = (String) context.getAttachment(RpcConstants.INTERNAL_KEY_ROUTER_RECORD);
record = record == null ? routerName : record + ">" + routerName;
context.setAttachment(RpcConstants.INTERNAL_KEY_ROUTER_RECORD, record);
}
}
子类必须实现 route 方法,该方法的参数是 SofaRequest 和一个 ProviderInfo List。然后,返回值是筛选过的 ProviderInfo List。即用户可以在自定义的 Router 中筛选 Router 使用。负载均衡会从用户的 ProviderInfo List 中选择一个 ProviderInfo 进行调用。
路由顺序按照 Extension 注解的 order 进行从小到大排序。
同时, SOFA 上下文 RpcInternalContext 会记录此次调用的路径,也就是路由名字。
而目前框架中有 3 个实现类:
- DirectUrlRouter 直连路由,needToLoad 判断条件是客户端是否设置了只来路由。路由规则是:从地址保持器中选取直连地址,然后添加到 List 中。
- ExcludeRouter 要排除的过滤器,目前暂没有定义。用户可自己扩展。
- RegistryRouter 从注册中心获取地址进行路由。needToLoad 条件是:如果没有设置直连地址且从注册中心订阅服务。路由规则:从地址保持器中获取默认的(注册中心)服务列表,并添加进 List 返回。
其中,这个地址保持/管理器是什么鬼呢?
每个客户端都有一个地址管理器 —— AddressHolder。管理着服务的分组。SingleGroupAddressHolder 是 AddressHolder 的具体实现类,也是通过扩展机制实现的。他是一个只支持单个分组的地址选择器(额外存一个直连分组)。
他内部有 2 个 List, 一个是直连地址列表,另一个是注册中心的地址列表。
在 Cluster 初始化的时候,会先初始化 routerChain Router 链,该实例中包含了一个 Router 数组,用于保存路由实例。
同时还会初始化服务端列表,即调用 consumerBootstrap.subscribe() 方法,该方法在 DefaultConsumerBootstrap 中实现如下:
- 如果直连地址(逗号或者分号分割)不是空,则返回一个包装了直连地址的 List
。 - 如果直连地址是空的,则从注册中心获取服务列表。
得到服务列表后,则添加到地址管理器中。同时向事件总线丢一个 ProviderInfoUpdateAllEvent 事件。包括建立长连接——也就是初始化 RpcClient,在调用 updateAllProviders 方法后,会异步的在一个 initPool 线程池中启动多个线程初始化长连接。
好了,现在地址管理器里面已经有连接了。
当客户端调用的时候,也就是 doInvoke 方法,会先从从 routerChain 中获取过滤后的 List
以上,就是自定义路由寻址,地址管理器的实现原理。
总结
SOFA 为框架用户定义了 Router ,用户可以实现 route 方法,并在该方法中实现过滤策略,从而返回一个用户设置的服务列表,值得注意的是,用户需要定义 order 属性,因为 Router 是从小到大排序的,顺序对于整体逻辑来说非常重要。
和 Router 息息相关的还有地址管理器 —— AddressHolder,该类会管理服务列表,客户端在初始化的时候,会将地址都保存到 AddressHolder 中,在之后的负载均衡选择服务的时候,会从地址管理器中获取服务列表(已经路由过滤的)进行选择。
还有一点需要注意:地址管理器每次 update 的时候,会全量更新连接管理器。如果有新增的服务,就会建立长连接。