今天来学习下sofa rpc中的过滤器链的设计方式，与传统servlet中的filter不太一样，值得学习下
客户端消费时序，只画了涉及filter部分
sofa-client-filter.jpg
服务消费端引用ConsumerConfig.refer()代码如下：
public T refer() {
if (consumerBootstrap == null) {
//创建消费启动类
consumerBootstrap = Bootstraps.from(this);
}
//调用一个消费启动类发布引用
return consumerBootstrap.refer();
}
调用ConsumerBootstrap.refer()，真正调用的是父类DefaultConsumerBootstrap.refer()，然后里面会创建FailoverCluster对象并初始化，执行链Filter就是在其父类AbstractCluster.init()内构造的，返回代理类，此代理类生成见【SOFA RPC客户端源码解析之客户端生成代理类】章节说明， 核心代码如下：
try {
// build cluster
cluster = ClusterFactory.getCluster(this);
// build listeners
consumerConfig.setConfigListener(buildConfigListener(this));
consumerConfig.setProviderInfoListener(buildProviderInfoListener(this));
// init cluster 今天主要解析此构造消费端Filter执行链逻辑
cluster.init();
// 构造Invoker对象
proxyInvoker = buildClientProxyInvoker(this);
// 创建代理类
proxyIns = (T) ProxyFactory.buildProxy(consumerConfig.getProxy(), consumerConfig.getProxyClass(),
proxyInvoker);
} catch (Exception e) {
if (cluster != null) {
cluster.destroy();
cluster = null;
}
consumerConfig.setConfigListener(null);
consumerConfig.setProviderInfoListener(null);
cnt.decrementAndGet(); // 发布失败不计数
if (e instanceof SofaRpcRuntimeException) {
throw (SofaRpcRuntimeException) e;
} else {
throw new SofaRpcRuntimeException("Build consumer proxy error!", e);
}
}
AbstractCluster.init()核心代码如下
if (initialized) { // 已初始化
return;
}
// 构造Router链
routerChain = RouterChain.buildConsumerChain(consumerBootstrap);
// 负载均衡策略 考虑是否可动态替换？
loadBalancer = LoadBalancerFactory.getLoadBalancer(consumerBootstrap);
// 地址管理器
addressHolder = AddressHolderFactory.getAddressHolder(consumerBootstrap);
// 连接管理器
connectionHolder = ConnectionHolderFactory.getConnectionHolder(consumerBootstrap);
// 构造Filter链,最底层是调用过滤器
this.filterChain = FilterChain.buildConsumerChain(this.consumerConfig,
new ConsumerInvoker(consumerBootstrap));
if (consumerConfig.isLazy()) { // 延迟连接
if (LOGGER.isInfoEnabled(consumerConfig.getAppName())) {
LOGGER.infoWithApp(consumerConfig.getAppName(), "Connection will be initialized when first invoke.");
}
}
// 启动重连线程
connectionHolder.init();
try {
// 得到服务端列表
List<ProviderGroup> all = consumerBootstrap.subscribe();
if (CommonUtils.isNotEmpty(all)) {
// 初始化服务端连接（建立长连接)
updateAllProviders(all);
}
} catch (SofaRpcRuntimeException e) {
throw e;
} catch (Throwable e) {
throw new SofaRpcRuntimeException("Init provider's transport error!", e);
}
// 启动成功
initialized = true;
// 如果check=true表示强依赖
if (consumerConfig.isCheck() && !isAvailable()) {
throw new SofaRpcRuntimeException("The consumer is depend on alive provider " +
"and there is no alive provider, you can ignore it " +
"by ConsumerConfig.setCheck(boolean) (default is false)");
}
下面来讲下今天主题构造Filter
// 构造Filter链,最底层是调用过滤器
this.filterChain = FilterChain.buildConsumerChain(this.consumerConfig,
new ConsumerInvoker(consumerBootstrap));
FilterChain.buildConsumerChain如下
public static FilterChain buildConsumerChain(ConsumerConfig<?> consumerConfig, FilterInvoker lastFilter) {
/*
* 例如自动装载扩展 A(a),B(b),C(c) filter=[-a,d] filterRef=[new E, new Exclude(b)]
* 逻辑如下：
* 1.解析config.getFilterRef()，记录E和-b
* 2.解析config.getFilter()字符串，记录 d 和 -a,-b
* 3.再解析自动装载扩展，a,b被排除了，所以拿到c,d
* 4.对c d进行排序
* 5.拿到C、D实现类
* 6.加上自定义，返回C、D、E
*/
// 用户通过自己new实例的方式注入的filter，优先级高
List<Filter> customFilters = consumerConfig.getFilterRef() == null ?
new ArrayList<Filter>() : new CopyOnWriteArrayList<Filter>(consumerConfig.getFilterRef());
// 先解析是否有特殊处理
HashSet<String> excludes = parseExcludeFilter(customFilters);
// 准备数据：用户通过别名的方式注入的filter，需要解析
List<ExtensionClass<Filter>> extensionFilters = new ArrayList<ExtensionClass<Filter>>();
List<String> filterAliases = consumerConfig.getFilter(); //
if (CommonUtils.isNotEmpty(filterAliases)) {
for (String filterAlias : filterAliases) {
if (startsWithExcludePrefix(filterAlias)) { // 排除用的特殊字符
excludes.add(filterAlias.substring(1));
} else {
ExtensionClass<Filter> filter = EXTENSION_LOADER.getExtensionClass(filterAlias);
if (filter != null) {
extensionFilters.add(filter);
}
}
}
}
// 解析自动加载的过滤器
if (!excludes.contains(StringUtils.ALL) && !excludes.contains(StringUtils.DEFAULT)) { // 配了-*和-default表示不加载内置
for (Map.Entry<String, ExtensionClass<Filter>> entry : CONSUMER_AUTO_ACTIVES.entrySet()) {
if (!excludes.contains(entry.getKey())) {
extensionFilters.add(entry.getValue());
}
}
}
excludes = null; // 不需要了
// 按order从小到大排序
if (extensionFilters.size() > 1) {
Collections.sort(extensionFilters, new OrderedComparator<ExtensionClass<Filter>>());
}
List<Filter> actualFilters = new ArrayList<Filter>();
for (ExtensionClass<Filter> extensionFilter : extensionFilters) {
actualFilters.add(extensionFilter.getExtInstance());
}
// 加入自定义的过滤器
actualFilters.addAll(customFilters);
return new FilterChain(actualFilters, lastFilter, consumerConfig);
}
上面一坨就不细看了，看最后一行代码 return new FilterChain(actualFilters, lastFilter, consumerConfig);他返回一个FilterChain对象，并把filters列表及lastFilter但此lastFilter为FilterInvoker对象传进行
下面来看new FilterChain
// 调用过程外面包装多层自定义filter
// 前面的过滤器在最外层
invokerChain = lastInvoker;
if (CommonUtils.isNotEmpty(filters)) {
loadedFilters = new ArrayList<Filter>();
for (int i = filters.size() - 1; i >= 0; i--) {
try {
Filter filter = filters.get(i);
if (filter.needToLoad(invokerChain)) {
invokerChain = new FilterInvoker(filter, invokerChain, config);
// cache this for filter when async respond
loadedFilters.add(filter);
}
} catch (Exception e) {
LOGGER.error("Error when build filter chain", e);
throw new SofaRpcRuntimeException("Error when build filter chain", e);
}
}
}
此构造函数是for循环Filter，并new FilterInvoker来包装，入参Filter,上一个invokerChain（FilterInvoker实例），这种做法最后拿到的invokerChain就是一个链表式数据结构，里面有一个FilterInvoker引用及Filter，可以看到下面代码判断nextFilter不为null，就调用的是Filter.invoke()，最后调用的是FilterInvoker.invoker()
if (nextFilter == null && invoker == null) {
throw new SofaRpcException(RpcErrorType.SERVER_FILTER, "Next filter or invoker is null!");
}
return nextFilter == null ?
invoker.invoke(request) :
nextFilter.invoke(invoker, request);
相当每次调Filter的时候其实是先调的FilterInvoker中的invoker，invoker中再调Filter
网友评论