Dubbo中Filter过滤器，拦截器的实现原理，实现自定义的Filter过滤器

Posted 2021-12-16 Leo Han

我们知道Dubbo中大部分的实现类加载都是通过SPI实现，同样Dubbo也提供了Filter机制，这个部分研究下怎么实现了，是怎样的一个调用逻辑。
首先我们看下Dubbo中Filter的定义：

 * Filter Chain in 3.x
 *
 *                                          -> Filter -> Invoker
 *
 * Proxy -> ClusterFilter -> ClusterInvoker -> Filter -> Invoker
 *
 *                                          -> Filter -> Invoker
@SPI
public interface Filter extends BaseFilter 

Dubbo中的filter可以在provider和consumer工作，而其注入的时机，我们需要回忆下之前说的Dubbo中的SPI机制实现，自定义对接SPI

我们知道，当我们加载Protocol的时候，如果有WrapperClass的话，先加载Wrapper，多个wrapper会嵌套，然后里面在嵌套一个真实的需要加载的类
同样对于Filter过滤器是基于Wrapper实现的

服务端

在Dubbo中默认的Protocol的Wrapper有下面几个：

ProtocolFilterWrapper
ProtocolListenerWrapper
QosProtocolWrapper

这里我们说下ProtocolFilterWrapper这里就是Dubbo中Filter过滤器实现的地方：

@Activate(order = 100)
public class ProtocolFilterWrapper implements Protocol 

    private final Protocol protocol;
    private static final FilterChainBuilder builder
            = ExtensionLoader.getExtensionLoader(FilterChainBuilder.class).getDefaultExtension();

    public ProtocolFilterWrapper(Protocol protocol) 
        if (protocol == null) 
            throw new IllegalArgumentException("protocol == null");
        
        this.protocol = protocol;
    

    @Override
    public int getDefaultPort() 
        return protocol.getDefaultPort();
    

    @Override
    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException 
        if (UrlUtils.isRegistry(invoker.getUrl())) 
            return protocol.export(invoker);
        
        return protocol.export(builder.buildInvokerChain(invoker, SERVICE_FILTER_KEY, CommonConstants.PROVIDER));
    
.....

而这里在export的时候，如果不是注册中心方式，会通过 FilterChainBuilder来构造一个FilterChainNode(注册中心方式的话，后面会修改url中的protocol，最后还是走这个逻辑)，FilterChainBuilder的Dubbo默认实现为DefaultFilterChainBuilder:

public <T> Invoker<T> buildInvokerChain(final Invoker<T> originalInvoker, String key, String group) 
        Invoker<T> last = originalInvoker;
        List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(originalInvoker.getUrl(), key, 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 FilterChainNode<>(originalInvoker, next, filter);
            
        

        return last;
    

这里通过ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(originalInvoker.getUrl(), key, group);会获取到一个Filter的集合列表，这里的key和group分别为：
SERVICE_FILTER_KEY, CommonConstants.PROVIDER
表示的是服务端的过滤和分组。
获取到Filter列表之后，构造了一个FilterChainNode链表，并返回链表的头部，注意这个FilterChainNode也是一个Invoker

这样在服务端我们进行export的暴露服务的时候，生成的是一个FilterChainNode的Invoker链表，而这里的Filter加载的顺序，还是之前基于SPI的分析，是根据Activate注解中order的顺序，从大到小顺序排序。
我们看下Dubbo中默认的Filter有哪些：

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
classloader=org.apache.dubbo.rpc.filter.ClassLoaderFilter
context=org.apache.dubbo.rpc.filter.ContextFilter
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
tps=org.apache.dubbo.rpc.filter.TpsLimitFilter

返回的Invoker是一个FilterChainNode，我们看下其invoke方法时怎么执行的：

public Result invoke(Invocation invocation) throws RpcException 
            Result asyncResult;
            try 
                asyncResult = filter.invoke(nextNode, invocation);
             catch (Exception e) 
                if (filter instanceof ListenableFilter) 
                    ListenableFilter listenableFilter = ((ListenableFilter) filter);
                    try 
                        Filter.Listener listener = listenableFilter.listener(invocation);
                        if (listener != null) 
                            listener.onError(e, originalInvoker, invocation);
                        
                     finally 
                        listenableFilter.removeListener(invocation);
                    
                 else if (filter instanceof FILTER.Listener) 
                    FILTER.Listener listener = (FILTER.Listener) filter;
                    listener.onError(e, originalInvoker, invocation);
                
                throw e;
             finally 

            
            return asyncResult.whenCompleteWithContext((r, t) -> 
                if (filter instanceof ListenableFilter) 
                    ListenableFilter listenableFilter = ((ListenableFilter) filter);
                    Filter.Listener listener = listenableFilter.listener(invocation);
                    try 
                        if (listener != null) 
                            if (t == null) 
                                listener.onResponse(r, originalInvoker, invocation);
                             else 
                                listener.onError(t, originalInvoker, invocation);
                            
                        
                     finally 
                        listenableFilter.removeListener(invocation);
                    
                 else if (filter instanceof FILTER.Listener) 
                    FILTER.Listener listener = (FILTER.Listener) filter;
                    if (t == null) 
                        listener.onResponse(r, originalInvoker, invocation);
                     else 
                        listener.onError(t, originalInvoker, invocation);
                    
                
            );
        

可以看到，调用的是filter.invoke(nextNode, invocation);并且会把下一个节点传递过去，而开始创建的那个filer中，nextNode就是原始的Invoker，这样就会形成一个调用链，即：最先开始调用的都是Filter的逻辑，Filter都执行完之后没有问题，执行原始的Invoker

我们举个ExecuteLimitFilter为例：

public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException 
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();
        int max = url.getMethodParameter(methodName, EXECUTES_KEY, 0);
        if (!RpcStatus.beginCount(url, methodName, max)) 
            throw new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION,
                    "Failed to invoke method " + invocation.getMethodName() + " in provider " +
                            url + ", cause: The service using threads greater than <dubbo:service executes=\\"" + max +
                            "\\" /> limited.");
        

        invocation.put(EXECUTE_LIMIT_FILTER_START_TIME, System.currentTimeMillis());
        try 
            return invoker.invoke(invocation);
         catch (Throwable t) 
            if (t instanceof RuntimeException) 
                throw (RuntimeException) t;
             else 
                throw new RpcException("unexpected exception when ExecuteLimitFilter", t);
            
        
    

可以看到这里在最终执行之前，会判断当前执行的方法正在执行的数量，如果大于给定的值，这里就会抛出异常，否则继续向后执行。

消费端

对于消费端，如果不是注册中心方式，和服务提供端完全一致，如果是基于注册中心的，这时候消费端和服务端则不一样了，消费端调用的是Protocol.refer，而这个方法返回的是一个Cluster，我们实现的几个Cluster都是基于AbstractCluster，而基于前面的分析Dubbo消费端启动流程、处理逻辑，方法调用实现,Cluster会和服务目录进行join之后返回一个Invoker,这里AbstractCluster.join实现如下：

 public <T> Invoker<T> join(Directory<T> directory) throws RpcException 
        if (directory instanceof StaticDirectory) 
            return doJoin(directory);
        
        return buildClusterInterceptors(doJoin(directory), directory.getUrl().getParameter(REFERENCE_INTERCEPTOR_KEY));
    

注册中心的Directory都不是StaticDirectory，会走第二个逻辑，而第二个逻辑则是构建Filter链的关键：

private <T> Invoker<T> buildClusterInterceptors(AbstractClusterInvoker<T> clusterInvoker, String key) 
        AbstractClusterInvoker<T> last = buildInterceptorInvoker(new ClusterFilterInvoker<>(clusterInvoker));
        if (Boolean.parseBoolean(ConfigurationUtils.getProperty(CLUSTER_INTERCEPTOR_COMPATIBLE_KEY, "false"))) 
            return build27xCompatibleClusterInterceptors(clusterInvoker, last);
        
        return last;
    
public ClusterFilterInvoker(AbstractClusterInvoker<T> invoker) 
            List<FilterChainBuilder> builders = ExtensionLoader.getExtensionLoader(FilterChainBuilder.class).getActivateExtensions();
            if (CollectionUtils.isEmpty(builders)) 
                filterInvoker = invoker;
             else 
                ClusterInvoker<T> tmpInvoker = invoker;
                for (FilterChainBuilder builder : builders) 
                    tmpInvoker = builder.buildClusterInvokerChain(tmpInvoker, REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
                
                filterInvoker = tmpInvoker;
            
        

        @Override
        public Result invoke(Invocation invocation) throws RpcException 
            return filterInvoker.invoke(invocation);
        

可以看到，这里生成了一个ClusterFilterInvoker，而在ClusterFilterInvoker中通过 FilterChainBuilder.buildClusterInvokerChain.生成了ClusterFilterChainNode链，而Dubbo中默认FilterChainBuilder就是DefaultFilterChainBuilder:

public <T> ClusterInvoker<T> buildClusterInvokerChain(final ClusterInvoker<T> originalInvoker, String key, String group) 
        ClusterInvoker<T> last = originalInvoker;
        List<ClusterFilter> filters = ExtensionLoader.getExtensionLoader(ClusterFilter.class).getActivateExtension(originalInvoker.getUrl(), key, group);
        if (!filters.isEmpty()) 
            for (int i = filters.size() - 1; i >= 0; i--) 
                final ClusterFilter filter = filters.get(i);
                final Invoker<T> next = last;
                last = new ClusterFilterChainNode<>(originalInvoker, next, filter);
            
        
        return last;
    

这里除了生成Filter链还会判断是否需要支持ClusterInterceptor,开启条件是dubbo.application.cluster.interceptor.compatible=true
但是请注意这里的Filter类型是ClusterFilter,返回的是一个ClusterFilterChainNode。

然后根据我们之前的研究Dubbo中服务注册与发现实现原理，消费端基于服务注册中心回去注册中心中获取对应的注册服务的的信息，然后进行初始化实际的Invoker，在ServiceDiscoveryRegistryDirectory中会进行实际的Invoker的初始化，这时候获取的protocol就是实际的protocl（实际上就是和服务端的一样，是一个Wrapper），而获取的这个protocol也是一个wrapper，还是和服务端一样的逻辑。
从这里能够看出，可能的处理链路如下：

Proxy -> Filter -> Invoker
Proxy -> ClusterFilter -> ClusterInvoker -> Filter -> Invoker

第一种一般在没有配置中心的时候，获取到的Protocol就是一个Wrapper的Protocol，这其中就有Filter的包装类

第二种一般是配有配置中心的情况，这个时候首先生成的是ClusterFilter然后从配置中心获取配置之后再次调用实际的Protocol进行处理，这时候获取的和第一种一样也是一个Wrapper的Protocol。

实现自定义Filter

如果我们要想实现自定义的Filter，那么我们可进行如下处理：

1. 实现Filter接口,且接口需要加上Activate注解，不加上该注解Filter不会被激活
2. 在当前的classpath下建立META-INF/dubbo/internal/、META-INF/dubbo/、META-INF/services/其中一个文件夹就行，然后在该目录下建立一个文件，名称为：org.apache.dubbo.rpc.Filter,内容为：

xxx=my.custom.XxxFilter

完成。