Dubbo源码分析系列之-深入Dubbo扩展机制

分手后的思念是犯贱 2021-10-15 01:45 325阅读 0赞

> 导语：  
>   在之前的博客中分析过Java的SPI机制，其实Dubbo的扩展点加载机制也是从JDK表中的SPI(Service Provider Interface)机制中开发而来，只不过在原生的基础上做了发现机制的增强处理。改进了如下的三个问题
> 
>  *  JDK的SPI机制会一次性的实例化所有的扩展点，也就是说数据一种饿汉式加载，在初始化的时候消耗比较大，但是有些资源被加载之后可能很少使用，所以就导致了资源的消耗问题。
>  *  如果扩展点加载失败，JDK不会获取扩展点名称，导致排查问题效率低下。
>  *  增加了Spring的对于扩展点的IOC以及AOP的支持操作。也就是说一个扩展点可以直接通过setter方式注入其他的扩展点。

### 扩展点配置 ###

了解完基本的问题之后就来看看关于JavaSPI机制以及Dubbo的SPI机制的一些基本约定，对于JavaSPI来说是将扩展点的配置文件放到META\_INF/目录下面并且与扩展接口同名的文件中写入对应的扩展类。而在Dubbo中在META\_INF/dubbo/接口全类名。文件的内容是k-v的形式，配置名=扩展类全类名，多个实现类之间使用的是换行符分隔，也就是文件中所有的配置形式都是key-value的形式。例如对于Dubbo协议的扩展。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]

dubbo=org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol

在配置文件中会指定对应的配置标签。

### 扩展点特性 ###

###### 扩展点包装 ######

在对于扩展点的使用的时候首先要对其进行包装，将其包装成一个比较完美的类。这就用到了装饰者模式，对于扩展点进行自动包装。Dubbo对于这些类的包装都是通过xxxWrapper中来进行包装。在ExtensionLoader在加载扩展点时，如果加载到扩展点有拷贝构造函数，则判断为扩展点的Wrapper，这个地方使用到了原型模式，从一个原型复制很多的克隆实例。这些设计模式在后期详细分析源码的时候都会提及到。这里就拿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;
        }
    
    
    
        private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) { 
            Invoker<T> last = invoker;
            List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.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 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();
                        }
                    };
                }
            }
    
            return new CallbackRegistrationInvoker<>(last, filters);
        }
    
        @Override
        public int getDefaultPort() { 
            return protocol.getDefaultPort();
        }
    
        @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));
        }
    
        @Override
        public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { 
            if (REGISTRY_PROTOCOL.equals(url.getProtocol())) { 
                return protocol.refer(type, url);
            }
            return buildInvokerChain(protocol.refer(type, url), REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
        }
    
        @Override
        public void destroy() { 
            protocol.destroy();
        }
    
        /** * Register callback for each filter may be better, just like {@link java.util.concurrent.CompletionStage}, each callback * registration generates a new CompletionStage whose status is determined by the original CompletionStage. * * If bridging status between filters is proved to not has significant performance drop, consider revert to the following commit: * https://github.com/apache/dubbo/pull/4127 */
        static class CallbackRegistrationInvoker<T> implements Invoker<T> { 
    
            private final Invoker<T> filterInvoker;
            private final List<Filter> filters;
    
            public CallbackRegistrationInvoker(Invoker<T> filterInvoker, List<Filter> filters) { 
                this.filterInvoker = filterInvoker;
                this.filters = filters;
            }
    
            @Override
            public Result invoke(Invocation invocation) throws RpcException { 
                Result asyncResult = filterInvoker.invoke(invocation);
    
                asyncResult = asyncResult.whenCompleteWithContext((r, t) -> { 
                    for (int i = filters.size() - 1; i >= 0; i--) { 
                        Filter filter = filters.get(i);
                        // onResponse callback
                        if (filter instanceof ListenableFilter) { 
                            Filter.Listener listener = ((ListenableFilter) filter).listener();
                            if (listener != null) { 
                                if (t == null) { 
                                    listener.onResponse(r, filterInvoker, invocation);
                                } else { 
                                    listener.onError(t, filterInvoker, invocation);
                                }
                            }
                        } else { 
                            filter.onResponse(r, filterInvoker, invocation);
                        }
                    }
                });
                return asyncResult;
            }
    
            @Override
            public Class<T> getInterface() { 
                return filterInvoker.getInterface();
            }
    
            @Override
            public URL getUrl() { 
                return filterInvoker.getUrl();
            }
    
            @Override
            public boolean isAvailable() { 
                return filterInvoker.isAvailable();
            }
    
            @Override
            public void destroy() { 
                filterInvoker.destroy();
            }
        }
    }

通过代码我们会发现Wrapper类同样实现了扩展点接口，但是Wrapper并不是真正的实现。它的主要作用是用于从Extensionloader返回扩展点的时候，对真正的扩展点进行包装。也就是在其中buildInvokerChain()所实现的内容，也就是说从ExtensionLoader中返回的实际上是Wrapper类的实例，Wrapper持有实际的扩展点实现类。  
  当然除了上面所展示的ProtocolFilterWrapper类还有其他的Wrapper，也就是说扩展点的包装类可以有多个，也可以根据需要继续增加，通过Wrapper类可以把所有扩展点公共逻辑移植到Wrapper中，新增加的Wrapper在所有的扩展点上添加了逻辑，有点像是AOP，Wrapper实际上是对扩展点的代理。

### 扩展点自动装配 ###

在前面提到了一个装饰者模式，在这里扩展点自动装配所使用的就是装配者模式。加载扩展点时会自动注入依赖的扩展点，加载扩展点的时候扩展点实现类的成员如果为其他扩展点类型ExtensionLoader会进行自动注入依赖的扩展点。ExtensionLoader通过扫描扩展点实现类的所有setter方法来判断其成员类型。也就是说ExtensionLoader会执行扩展点的拼装操作。

例如我们现在有个汽车工厂，其中有两个制造车间  
制造轮子

public interface MarkerWheel{ 
    	Wheel makeWheel();
    }

制造引擎

public interface MarkerEngine{ 
    	Engine makerEngine();
    }

MarkerWheel 的实现类

public class RealMarkerWheel implements MarkerWheel{ 
    	MarkerWheel markerWheel;
    	
    	public setWheelMaker(MarkerWheel markerWheel){ 
    		this.markerWheel = markerWheel;
    	}
    
    	public Car makeCar(){ 
    		Wheel wheel = markerWheel.makeWheel();
    
    		return new CarFactory(wheel,......)
    	}
    }

当ExtensionLoader加载的Car的扩展点实现的时候，调用setWheelMaker方法如果MarkerWheel也是扩展点则会注入MarkerWheel的实现并且实现装配。  
  在这里所带来的一个问题就是ExtensionLoader要注入依赖扩展点的时候，如果决定使用哪个依赖扩展点的实现，也就是说如果有高中低三种类型的轮胎，在组装汽车的时候应该使用哪一种轮胎进行组装。

### 扩展点自适应 ###

首先ExtensionLoader注入的依赖扩展点是一个Adaptive实例，直到扩展点执行的时候才会决定调用哪个实现。Dubbo使用URL对象或者使用Key-Value 的方式进行传递参数的配置信息。扩展点方法调用都会有URL参数或者是类似的操作成员。  
  这样依赖的扩展点可以从URL中获取到配置的信息，所有的扩展点配置好自己的key后，配置信息从URL上从最外层传入，URL在配置传递的过程中是一条总线的服务。

例如我们现在有个汽车工厂，其中有两个制造车间  
制造轮子

public interface MarkerWheel{ 
    	Wheel makeWheel(URL url);
    }

制造引擎

public interface MarkerEngine{ 
    	Engine makerEngine(URL url);
    }

MarkerWheel 的实现类

public class RealMarkerWheel implements MarkerWheel{ 
    	MarkerWheel markerWheel;
    	
    	public setWheelMaker(MarkerWheel markerWheel){ 
    		this.markerWheel = markerWheel;
    	}
    
    	public Car makeCar(URL url){ 
    		Wheel wheel = markerWheel.makeWheel(url);
    
    		return new CarFactory(wheel,......)
    	}
    }

当执行Wheel wheel = markerWheel.makeWheel(url);方法的时候，注入的Adaptive实例可以提取约定Key来决定使用哪个MarkerWheel实现来调用对应实现真正的markerWheel.makeWheel()方法，例如上面提到的高中低三个层次的轮子。对于Adaptive的实现逻辑来说是固定的。指定提取的就是URL的key，也就是可以代理在真实的实现类上，可以动态生成。

在Dubbo的ExtensionLoader的扩展点对应的Adaptive实现是在加载扩展点里动态生成，指定提取的URL的Key通过@Adaptive注解在方法上提供。例如在Dubbo中的Transport扩展点的代码

@SPI("netty")
    public interface Transporter { 
    
        /** * Bind a server. * * @param url server url * @param handler * @return server * @throws RemotingException * @see org.apache.dubbo.remoting.Transporters#bind(URL, ChannelHandler...) */
        @Adaptive({ Constants.SERVER_KEY, Constants.TRANSPORTER_KEY})
        Server bind(URL url, ChannelHandler handler) throws RemotingException;
    
        /** * Connect to a server. * * @param url server url * @param handler * @return client * @throws RemotingException * @see org.apache.dubbo.remoting.Transporters#connect(URL, ChannelHandler...) */
        @Adaptive({ Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
        Client connect(URL url, ChannelHandler handler) throws RemotingException;
    
    }

对于bind()方法，Adaptive实现先查找server Key，如果没有找到再找transport Key，来决定代理到哪个扩展点点。对于connect()方法也是类似。

### 扩展点自动激活 ###

在对扩展点的自动激活上，会看到Dubbo上有另外的注解@Activate，例如集合扩展点，Filter、InvokerListener、ExportListener、TelnetHandler、StatusChecker等等，同时可以加载多个实现类。这个时候就可以通过提供的自动激活来简化配置。例如

@Activate(group = PROVIDER, value = ACCESS_LOG_KEY)
    public class AccessLogFilter implements Filter { 
    
        private static final Logger logger = LoggerFactory.getLogger(AccessLogFilter.class);
    
        private static final String LOG_KEY = "dubbo.accesslog";

当然在使用@Activate注解的时候也可以不指定值，也可以指定单个值。

#### 注意 ####

*  1、在使用SPI配置的时候，所配置的META\_INF/目录是放在开发者的jar内，而不是dubbo本身的jar包内，Dubbo会全Classpath扫描所有的Jar包内的同名这个文件并进行合并操作。
 *  2、扩展点使用的是单一实例加载(也就是在扩展的时候需要保证线程安全)，缓存在ExtensionLoader中。

## 总结 ##

这篇博客结合源码以及官网的文档对于Dubbo的扩展机制进行深入的说明，提到了在Dubbo中的三个比较重要的注解，@SPI、@Adaptive、@Activate三个注解。在实际操作的时候也提到了它是对JavaSPI机制的扩展解决了JavaSPI机制存在的三个问题。基于Spring容器对于扩展点提供了IOC和AOP的功能。利用到了装配者模式，原型模式、代理模式以及装饰者模式等等。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]: /images/20211015/2652851cd51242ad9ff2bb7a72c6a825.png