AI 助理
备案控制台
开发者社区 Java技术进阶 文章 正文

Dubbo provider Filter链原理

2019-10-22 1782
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 开篇 在dubbo的使用过程中会在标签中会配置filter变量，但是filter具体如何生效却不是特别清楚，这篇文章就是针对Filter的加载过程进行下分析，尝试描述清楚过程。 在这篇文章中会尝试解释ProtocolFilterWrapper被调用过程，协议发布的时候都会走到ProtocolFilterWrapper，而这个类是Filter的加载入口，其核心方法在buildInvokerChain()当中。

开篇

 在dubbo的使用过程中会在标签中会配置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 调用链


  • 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的执行过程，两个问题在上面已经都提到并进行了一定的解释。


参考

文章标签：
应用服务中间件
Dubbo
缓存
关键词：
Dubbo原理
Dubbo provider
Dubbo filter
Dubbo provider filter
晴天哥
目录
相关文章
mikechen的互联网架构
|
1月前
|
负载均衡 监控 Dubbo
Dubbo 原理和机制详解（非常全面）
本文详细解析了 Dubbo 的核心功能、组件、架构设计及调用流程，涵盖远程方法调用、智能容错、负载均衡、服务注册与发现等内容。欢迎留言交流。关注【mikechen的互联网架构】，10年+BAT架构经验倾囊相授。
mikechen的互联网架构
176 5
Dubbo 原理和机制详解（非常全面）
zuozewei
|
7月前
|
负载均衡 监控 Dubbo
秒懂Dubbo接口（原理篇）
【4月更文挑战第25天】秒懂Dubbo接口（原理篇）
zuozewei
248 3
秒懂Dubbo接口（原理篇）
服务端技术栈
|
4月前
|
负载均衡 Dubbo 应用服务中间件
Dubbo服务调用过程原理
该文章主要介绍了Dubbo服务调用过程的原理，包括服务调用的主要阶段和服务调用的具体步骤。
服务端技术栈
54 0
Dubbo服务调用过程原理
服务端技术栈
|
4月前
|
缓存 Dubbo Java
Dubbo服务消费者启动与订阅原理
该文章主要介绍了Dubbo服务消费者启动与订阅的原理，包括服务消费者的启动时机、启动过程以及订阅和感知最新提供者信息的方式。
服务端技术栈
64 0
Dubbo服务消费者启动与订阅原理
服务端技术栈
|
4月前
|
Dubbo 网络协议 Java
深入掌握Dubbo服务提供者发布与注册原理
该文章主要介绍了Dubbo服务提供者发布与注册的原理，包括服务发布的流程、多协议发布、构建Invoker、注册到注册中心等过程。
服务端技术栈
58 0
深入掌握Dubbo服务提供者发布与注册原理
服务端技术栈
|
4月前
|
负载均衡 Dubbo Java
Dubbo服务Spi机制和原理
该文章主要介绍了Dubbo中的SPI（Service Provider Interface）机制和原理，包括SPI的基本概念、Dubbo中的SPI分类以及SPI机制的实现细节。
服务端技术栈
83 0
Dubbo服务Spi机制和原理
服务端技术栈
|
4月前
|
缓存 负载均衡 Dubbo
Dubbo服务集群容错原理（重要）
该文章主要介绍了Dubbo服务集群容错的原理，包括集群容错技术的概念、Dubbo中使用的集群容错技术种类及其原理。
服务端技术栈
55 0
服务端技术栈
|
4月前
|
负载均衡 Dubbo 算法
Dubbo服务负载均衡原理
该文章主要介绍了Dubbo服务负载均衡的原理，包括Dubbo中负载均衡的实现位置、为什么需要负载均衡机制、Dubbo支持的负载均衡算法以及随机负载均衡策略的源码分析。
服务端技术栈
44 0
磊哥聊Java
|
6月前
|
负载均衡 Dubbo Java
哈啰面试：说说Dubbo运行原理？
哈啰面试：说说Dubbo运行原理？
磊哥聊Java
52 0
哈啰面试：说说Dubbo运行原理？
老板这功能得加钱
|
7月前
|
负载均衡 算法
Dubbo-负载均衡原理解析(1)，一个本科渣渣是怎么逆袭从咸鱼到Offer收割机的
Dubbo-负载均衡原理解析(1)，一个本科渣渣是怎么逆袭从咸鱼到Offer收割机的
老板这功能得加钱
61 6

Java技术进阶

热门文章

最新文章

  • 1
    Spring-boot+Dubbo应用启停源码分析
  • 2
    Dubbo的Zookeeper单机配置和Zookeeper集群配置
  • 3
    Dubbo入门
  • 4
    高并发架构系列：如何从0到1设计一个类Dubbo的RPC框架
  • 5
    Apache Dubbo 被曝出“高危”远程代码执行漏洞
  • 6
    学习dubbo（四）： 启动时检查
  • 7
    Dubbo之——改造Dubbo，使其能够兼容Spring 4注解配置
  • 8
    Dubbo Consumer响应provider变化过程
  • 9
    Dubbo Triple 协议重磅升级：支持通过 HTTP 连通 Web 与后端微服务
  • 10
    Dubbo与Zookeeper、SpringMVC整合和使用（负载均衡、容错）（转）
  • 1
    SpringCloudAlibaba:3.1dubbo
    135
  • 2
    SpringCloudAlibaba:3.2dubbo的高级特性
    73
  • 3
    Spring Cloud Dubbo: 微服务通信的高效解决方案
    173
  • 4
    秒懂Dubbo接口（原理篇）
    249
  • 5
    SpringCloud和Dubbo有哪些区别
    144
  • 6
    微服务学习 | Springboot整合Dubbo+Nacos实现RPC调用
    933
  • 7
    Java从入门到精通：3.2.2分布式与并发编程——了解分布式系统的基本概念，学习使用Dubbo、Spring Cloud等分布式框架
    553
  • 8
    Dubbo管理控制台
    54
  • 9
    原来Dubbo中的缓存玩的这么花
    149
  • 10
    超越接口：探索Dubbo的泛化调用机制
    413

    • 相关课程

    更多
  • Dubbo + ZooKeeper 的服务发现最佳实践
  • Dubbo 3.0 服务治理最佳实践
  • Dubbo开源、自研、商业化三位一体战略解读
  • 基于Zookeeper、Dubbo构建互联网分布式基础架构
  • Dubbo分布式服务治理实战

    • 相关电子书

    更多
  • Dubbo开源现状与2.7规划
  • Dubbo分布式服务治理实战
  • 《Dubbo 3.0 前瞻》

    • 相关实验场景

    更多
  • 快速玩转Dubbo生态之一致性事务篇（Seata）
  • 快速玩转Dubbo生态之限流降级篇（Sentinel）
  • 快速上手使用Dubbo进行远程调用
  • Spring-cloud-stream-binder-rocketmq入门与实践
    • 下一篇
    fs.oss.accessKeyId和fs.oss.accessKeySecret。