Dubbo源码分析系列-Dubbo调用拦截器扩展

59 阅读 0 评论 39 点赞

我是靠谱客的博主辛勤月亮，最近开发中收集的这篇文章主要介绍Dubbo源码分析系列-Dubbo调用拦截器扩展，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

导语：
在之前的博客中说到了Dubbo的整体设计架构，以及协议扩展原理。这篇博客中主要介绍关于Dubbo协议的扩展。对于服务提供方和服务消费方调用过程的拦截，Dubbo本身的大多功能都能在它提供的拦截器扩展点上实现，每一次的远程调用这个拦截器都会被实现。

Filter(过滤器)在很多的框架中都用到过这个概念，基本上的作用都是比较相似的，调用前或者调用之后做一些通用的配置处理。当然对于Filter来说也可以有多个，也可以层层嵌套。在Dubbo中的Filter的概念与我们理解的类似，而且在Dubbo官方对于Filter也做了很多的扩展，下面我们会看到对于那些功能做了扩展。那么下面就来看看Dubbo中的Filter。

上的介绍中提拦截器的作用，在Dubbo中Filter拦截器的作用就是对于消费方或者服务提供方的调用过程的前置处理和后置处理。但是有的时候Dubbo原生的Filter并不能完全的满足开发的需要，就需要对于Filter进行扩展。在Dubbo中提供的SPI机制，并且对Filter进行了一些扩展。
Filter接口

@SPI
public interface Filter {
    /**
     * Does not need to override/implement this method.
     */
    Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException;

    /**
     * Filter itself should only be response for passing invocation, all callbacks has been placed into {@link Listener}
     *
     * @param appResponse
     * @param invoker
     * @param invocation
     * @return
     */
    @Deprecated
    default Result onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
        return appResponse;
    }

    interface Listener {

        void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation);

        void onError(Throwable t, Invoker<?> invoker, Invocation invocation);
    }

}

首先会看到在Filter接口中提供了两个方法和一个内置的监听接口。接口中定义了两个方法一个用来接收响应成功，一个用来处理失败响应。在实现接口的时候需要实现接口中的每一个方法。
Filter接口已知扩展类
在SPI配置文件中可以看到有如下的一些已知的扩展类

echo=org.apache.dubbo.rpc.filter.EchoFilter

generic=org.apache.dubbo.rpc.filter.GenericFilter

genericimpl=org.apache.dubbo.rpc.filter.GenericImplFilter

token=org.apache.dubbo.rpc.filter.TokenFilter

accesslog=org.apache.dubbo.rpc.filter.AccessLogFilter

activelimit=org.apache.dubbo.rpc.filter.ActiveLimitFilter

classloader=org.apache.dubbo.rpc.filter.ClassLoaderFilter

context=org.apache.dubbo.rpc.filter.ContextFilter

consumercontext=org.apache.dubbo.rpc.filter.ConsumerContextFilter

exception=org.apache.dubbo.rpc.filter.ExceptionFilter

executelimit=org.apache.dubbo.rpc.filter.ExecuteLimitFilter

deprecated=org.apache.dubbo.rpc.filter.DeprecatedFilter

compatible=org.apache.dubbo.rpc.filter.CompatibleFilter

timeout=org.apache.dubbo.rpc.filter.TimeoutFilter

下面就来以扩展ExceptionFilter为例说明一下Dubbo的Filter怎么实现扩展

@Activate(group = CommonConstants.PROVIDER)
public class ExceptionFilter extends ListenableFilter {

首先应该注意到的就是@Activate(group = CommonConstants.PROVIDER)这个注解，这个注解表示在什么时候激活，会看到这个拦截器在group为provider的时候被激活。并且这扩展类实现了ListenableFilter可监听拦截器。它也是作为一个拦截器而存在。实际上这个通过这个暴露的就是在Filter内部的接口Listener。而对于拦截器的核心实现其实就是通过实现内部的Listener来实现的。

public abstract class ListenableFilter implements Filter {

    protected Listener listener = null;

    public Listener listener() {
        return listener;
    }
}

拦截器是怎么生效的呢？
在ProtocolFilterWrapper类中有一个buildInvokerChain()方法，在这个方法中有如下的代码

List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);

通过类加载器进行加载到标注有@Activate标记的所有Filter列表。也就是前面提到的那些扩展并且这些扩展都是标注有group值的。
判断拦截器列表是否为空。

if (!filters.isEmpty()) {
    for (int i = filters.size() - 1; i >= 0; i--) {
         final Filter filter = filters.get(i);
         final Invoker<T> next = last;
         last = new Invoker<T>() {

           @Override
           public Class<T> getInterface() {
                return invoker.getInterface();
           }

           @Override
           public URL getUrl() {
                 return invoker.getUrl();
           }

           @Override
           public boolean isAvailable() {
                return invoker.isAvailable();
           }

           @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;
             }

              @Override
              public void destroy() {
                   invoker.destroy();
               }

               @Override
              public String toString() {
                   return invoker.toString();
              }
        };
     }
}

添加完成之后调用CallbackRegistrationInvoker<>(last, filters);回调函数将Invoker重新注册。

总结

在Dubbo中提供的所有的Filter都是基于这样的实现方式，只是对于每个Filter有每个Filter的功能点。在Dubbo中对于自定义的Filter与Dubbo原生的Filter的优先级。在Dubbo加载的时候首先加载的就是Dubbo内置的Filter，如果在项目中需要使用到其他的Filter也可以在原生的基础上进行扩展。