开篇
在dubbo的使用过程中会在<dubbo:service>标签中会配置filter变量，但是filter具体如何生效却不是特别清楚，这篇文章就是针对Filter的加载过程进行下分析，尝试描述清楚过程。
在这篇文章中会尝试解释ProtocolFilterWrapper被调用过程，协议发布的时候都会走到ProtocolFilterWrapper，而这个类是Filter的加载入口，其核心方法在buildInvokerChain()当中。
进而在buildInvokerChain()方法中通过ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension()获取所有的Filter（包括系统和自定义的Filter对象）的加载过程。
Filter调用链分析
- Dubbo的调用链如下图，调用顺序按照ProtocolListenerWrapper => ProtocolFilterWrapper，包含关系是 ProtocolListenerWrapper包含ProtocolFilterWrapper。
- 执行export()方法是会按照ProtocolListenerWrapper => ProtocolFilterWrapper的顺序执行，最终会调用ProtocolFilterWrapper的export()方法，进而进入buildInvokerChain()构造Filter链的逻辑。
InvocationChain构建链
Dubbo 调用链
- RegistryProtocol => DubboProtocol的调用链如下图。
Reistry-Dubbo协议调用链
- RegistryProtocol 中URL内容如下，所以Protocol@Adaptive访问RegistryProtocol对象。
registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?
application=dubbo-demo-api-provider&dubbo=2.0.2
&pid=50012®istry=zookeeper×tamp=1571727131732
- DubboProtocol 中URL如下，所以Protocol@Adaptive访问DubboProtocol对象。
dubbo://172.17.32.176:20880/org.apache.dubbo.demo.DemoService?
anyhost=true&application=dubbo-demo-api-provider&bind.ip=172.17.32.176
&bind.port=20880&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false
&interface=org.apache.dubbo.demo.DemoService&methods=sayHello
&pid=50012&release=&side=provider×tamp=1571727131742
- org.apache.dubbo.rpc.Protocol文件内容如下图，其中按照先后顺序是先filter后listener。
文件：
org.apache.dubbo.rpc.Protocol
内容：
filter=org.apache.dubbo.rpc.protocol.ProtocolFilterWrapper
listener=org.apache.dubbo.rpc.protocol.ProtocolListenerWrapper
- Protocol对应ExtensionLoader对象中cachedWrapperClasses包含filter对象和listener对象，且顺序先filter后listener。
Protocol对应ExtensionLoader
Filter调用链源码分析
- ProtocolListenerWrapper 包含一个Protocol类型的构造函数，会根据构造函数进行初始化。
- ProtocolFilterWrapper 包含一个Protocol类型的构造函数，会根据构造函数进行初始化。
public class ProtocolListenerWrapper implements Protocol {
private final Protocol protocol;
public ProtocolListenerWrapper(Protocol protocol) {
if (protocol == null) {
throw new IllegalArgumentException("protocol == null");
}
this.protocol = protocol;
}
}
public class ProtocolFilterWrapper implements Protocol {
private final Protocol protocol;
public ProtocolFilterWrapper(Protocol protocol) {
if (protocol == null) {
throw new IllegalArgumentException("protocol == null");
}
this.protocol = protocol;
}
}
- 适配器Protocol$Adaptive的export()方法会调用ExtensionLoader的getExtension()方法。
public class Protocol$Adaptive implements org.apache.dubbo.rpc.Protocol {
public org.apache.dubbo.rpc.Exporter export(org.apache.dubbo.rpc.Invoker arg0)
throws org.apache.dubbo.rpc.RpcException {
if (arg0 == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null");
if (arg0.getUrl() == null) throw new IllegalArgumentException(
"org.apache.dubbo.rpc.Invoker argument getUrl() == null");
org.apache.dubbo.common.URL url = arg0.getUrl();
String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
if(extName == null) throw new IllegalStateException(
"Failed to get extension (org.apache.dubbo.rpc.Protocol) name from url ("
+ url.toString() + ") use keys([protocol])");
org.apache.dubbo.rpc.Protocol extension = (org.apache.dubbo.rpc.Protocol)ExtensionLoader
.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension(extName);
return extension.export(arg0);
}
}
- getExtension()方法内部调用createExtension()创建扩展。
- createExtension()内部主要包括四个核心步骤。
- 步骤一：根据扩展名获取扩展类，getExtensionClasses().get(name)。
- 步骤二：创建扩展类的实例对象，clazz.newInstance()。
- 步骤三：根据set方法注入实例依赖的对象，injectExtension(instance)。
- 步骤四：创建包装类对象并注入扩展类实例对象，wrapperClass.getConstructor(type).newInstance(instance)。
- 在Protocol的ExtensionLoader的cachedWrapperClasses包括ProtocolFilterWrapper和ProtocolListenerWrapper。
- 继续分析ProtocolFilterWrapper类。
public class ExtensionLoader<T> {
public T getExtension(String name) {
if (StringUtils.isEmpty(name)) {
throw new IllegalArgumentException("Extension name == null");
}
if ("true".equals(name)) {
return getDefaultExtension();
}
final Holder<Object> holder = getOrCreateHolder(name);
Object instance = holder.get();
if (instance == null) {
synchronized (holder) {
instance = holder.get();
if (instance == null) {
instance = createExtension(name);
holder.set(instance);
}
}
}
return (T) instance;
}
private T createExtension(String name) {
// 根据扩展名获取扩展类clazz
Class<?> clazz = getExtensionClasses().get(name);
if (clazz == null) {
throw findException(name);
}
try {
T instance = (T) EXTENSION_INSTANCES.get(clazz);
if (instance == null) {
// 创建扩展类对象
EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance());
instance = (T) EXTENSION_INSTANCES.get(clazz);
}
// 注入instance对象，这里假设为DubboProtocol对象作为例子
// injectExtension方法内部遍历set方法并从上下文获取并set到对象当中
injectExtension(instance);
// 获取包装类WrapperClasses并挨个进行包装
// 包装类以instance的type作为构造函数的参数
Set<Class<?>> wrapperClasses = cachedWrapperClasses;
if (CollectionUtils.isNotEmpty(wrapperClasses)) {
for (Class<?> wrapperClass : wrapperClasses) {
instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));
}
}
return instance;
} catch (Throwable t) {
throw new IllegalStateException("Extension instance (name: " + name + ", class: " +
type + ") couldn't be instantiated: " + t.getMessage(), t);
}
}
private T injectExtension(T instance) {
if (objectFactory == null) {
return instance;
}
try {
for (Method method : instance.getClass().getMethods()) {
if (!isSetter(method)) {
continue;
}
if (method.getAnnotation(DisableInject.class) != null) {
continue;
}
Class<?> pt = method.getParameterTypes()[0];
if (ReflectUtils.isPrimitives(pt)) {
continue;
}
try {
// 获取set方法的属性property
String property = getSetterProperty(method);
// 这里的objectFactory为AdaptiveExtensionFactory
// objectFactory.getExtension执行
// SpiExtensionFactory的getExtension()方法
// 返回loader.getAdaptiveExtension()对象
// Protocol$Adaptive对象
Object object = objectFactory.getExtension(pt, property);
if (object != null) {
method.invoke(instance, object);
}
} catch (Exception e) {
}
}
} catch (Exception e) {
}
return instance;
}
}
- 补充一句，当injectExtension()的参数instance为RegistryProtocol，RegistryProtocol包含setProtocol()方法，执行反射调用method.invoke(instance, object)注入Protocol$Adaptive对象。比较抽象，需要好好debug理解
ProtocolFilterWrapper分析
- 导出Dubbo协议过程中ProtocolFilterWrapper的Protocol为DubboProtocol对象。
- 执行ProtocolFilterWrapper的export()的参数为buildInvokerChain()包装后包含过滤器的invoker对象。
- buildInvokerChain()方法包含两个核心步骤，获取Filter列表和组装Filter列表。
- 先关注Filter列表的过程，获取过程后面继续分析。
- 假设Filter链为"A,B,C"，那么实际串联后的顺序为A => B => C => Invoker。
- 串联的逻辑代码在注释中标注了，逻辑是通过next记录上一次对象并保存当前Filter当中。
public class ProtocolFilterWrapper implements Protocol {
private final Protocol protocol;
public ProtocolFilterWrapper(Protocol protocol) {
if (protocol == null) {
throw new IllegalArgumentException("protocol == null");
}
this.protocol = protocol;
}
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
// 保存引用，后续用于把真正的调用者保存到过滤器链的最后
Invoker<T> last = invoker;
// 获取系统和自定义的Filter过滤器
List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class)
.getActivateExtension(invoker.getUrl(), key, group);
if (!filters.isEmpty()) {
// 用过滤器包装真正的invoker，过滤器的包装顺序是从尾到头，按照顺序逆向包装。
for (int i = filters.size() - 1; i >= 0; i--) {
final Filter filter = filters.get(i);
// 每次把last对象设置为next，挂到当前Filter的next当中，实现串联。
final Invoker<T> next = last;
last = new Invoker<T>() {
// 只关注重点代码
@Override
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult;
try {
asyncResult = filter.invoke(next, invocation);
} catch (Exception e) {
// onError callback
if (filter instanceof ListenableFilter) {
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
listener.onError(e, invoker, invocation);
}
}
throw e;
}
return asyncResult;
}
};
}
}
return new CallbackRegistrationInvoker<>(last, filters);
}
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
if (REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
return protocol.export(invoker);
}
return protocol.export(buildInvokerChain(invoker, SERVICE_FILTER_KEY, CommonConstants.PROVIDER));
}
}
- Dubbo原生的filter定义在META-INF/dubbo/internal/com.alibaba.dubbo.rpc.filter文件。
echo=com.alibaba.dubbo.rpc.filter.EchoFilter
generic=com.alibaba.dubbo.rpc.filter.GenericFilter
genericimpl=com.alibaba.dubbo.rpc.filter.GenericImplFilter
token=com.alibaba.dubbo.rpc.filter.TokenFilter
accesslog=com.alibaba.dubbo.rpc.filter.AccessLogFilter
activelimit=com.alibaba.dubbo.rpc.filter.ActiveLimitFilter
classloader=com.alibaba.dubbo.rpc.filter.ClassLoaderFilter
context=com.alibaba.dubbo.rpc.filter.ContextFilter
consumercontext=com.alibaba.dubbo.rpc.filter.ConsumerContextFilter
exception=com.alibaba.dubbo.rpc.filter.ExceptionFilter
executelimit=com.alibaba.dubbo.rpc.filter.ExecuteLimitFilter
deprecated=com.alibaba.dubbo.rpc.filter.DeprecatedFilter
compatible=com.alibaba.dubbo.rpc.filter.CompatibleFilter
timeout=com.alibaba.dubbo.rpc.filter.TimeoutFilter
monitor=com.alibaba.dubbo.monitor.support.MonitorFilter
validation=com.alibaba.dubbo.validation.filter.ValidationFilter
cache=com.alibaba.dubbo.cache.filter.CacheFilter
trace=com.alibaba.dubbo.rpc.protocol.dubbo.filter.TraceFilter
future=com.alibaba.dubbo.rpc.protocol.dubbo.filter.FutureFilter
- Filter的类图是下图所示，其中接口Filter包含了SPI注解。
Filter类图
- 以EchoFilter作为例子，我们可以看到Filter的实现都带有@Activate注解，加载类的时候会把带有@Activate注解的类进行单独缓存。
@Activate(group = CommonConstants.PROVIDER, order = -110000)
public class EchoFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
if (inv.getMethodName().equals($ECHO) && inv.getArguments() != null && inv.getArguments().length == 1) {
return AsyncRpcResult.newDefaultAsyncResult(inv.getArguments()[0], inv);
}
return invoker.invoke(inv);
}
}
- ExtensionLoader的getActivateExtension()内部主要执行获取系统Filter和用户自定义Filter两步骤。
- 获取系统Filter当中，如果filter标签不带有"-default"字段，就会执行获取系统Filter对象。
- 获取系统Filter对象是从cachedActivates获取的，而cachedActivates是在ExtensionLoader加载扩展类的时候缓存带有@Activate的类，也就是@Activate的Filter会在加载扩展类的时被缓存。
- 系统Filter保存在exts并按照一定规则进行排序。
- 获取自定义Filter根据filter标签定义的Filter，加载过程中不加载带有排除符号"-"的Filter，在关键字"default"之前的Filter会被优先加载。
- 自定义Filter一般不带@Activate标签，如果带@Activate标签就会进入系统Filter的加载流程。
public class ExtensionLoader<T> {
public List<T> getActivateExtension(URL url, String key, String group) {
String value = url.getParameter(key);
return getActivateExtension(url, StringUtils.isEmpty(value) ? null : COMMA_SPLIT_PATTERN.split(value), group);
}
public List<T> getActivateExtension(URL url, String[] values, String group) {
List<T> exts = new ArrayList<>();
List<String> names = values == null ? new ArrayList<>(0) : Arrays.asList(values);
// 如果Filter中不带有"-default"字段，就会加载系统扩展Filter对象。
if (!names.contains(REMOVE_VALUE_PREFIX + DEFAULT_KEY)) {
// 获取所有的扩展类
getExtensionClasses();
for (Map.Entry<String, Object> entry : cachedActivates.entrySet()) {
String name = entry.getKey();
Object activate = entry.getValue();
String[] activateGroup, activateValue;
if (activate instanceof Activate) {
activateGroup = ((Activate) activate).group();
activateValue = ((Activate) activate).value();
} else if (activate instanceof com.alibaba.dubbo.common.extension.Activate) {
activateGroup = ((com.alibaba.dubbo.common.extension.Activate) activate).group();
activateValue = ((com.alibaba.dubbo.common.extension.Activate) activate).value();
} else {
continue;
}
if (isMatchGroup(group, activateGroup)
&& !names.contains(name)
&& !names.contains(REMOVE_VALUE_PREFIX + name)
&& isActive(activateValue, url)) {
exts.add(getExtension(name));
}
}
// 按照一定的比较规则进行排序
exts.sort(ActivateComparator.COMPARATOR);
}
// 加载用户自定义扩展Filter对象
List<T> usrs = new ArrayList<>();
for (int i = 0; i < names.size(); i++) {
String name = names.get(i);
// 带有排除符号"-"的Filter不加载
if (!name.startsWith(REMOVE_VALUE_PREFIX)
&& !names.contains(REMOVE_VALUE_PREFIX + name)) {
if (DEFAULT_KEY.equals(name)) {
// 在default之前的Filter会被系统Filter之前被加载
if (!usrs.isEmpty()) {
exts.addAll(0, usrs);
usrs.clear();
}
} else {
usrs.add(getExtension(name));
}
}
}
if (!usrs.isEmpty()) {
exts.addAll(usrs);
}
return exts;
}
}
- 总体来说，针对Filter加载的过程需要搞清楚两个问题，一个问题是ProtocolFilterWrapper对象的调用链路，另外一个是buildInvokerChain的执行过程，两个问题在上面已经都提到并进行了一定的解释。
