Dubbo插件化
Dubbo的插件化实现非常类似于原生的JAVA的SPI：它只是提供一种协议，并没有提供相关插件化实施的接口。用过的同学都知道，它有一种java原生的支持类：ServiceLoader，通过声明接口的实现类，在META-INF/services中注册一个实现类，然后通过ServiceLoader去生成一个接口实例，当更换插件的时候只需要把自己实现的插件替换到META-INF/services中即可。
SPI
Dubo Spi 和Java SPi的使用和对比
在Dubbo中，SPI是一个非常核心的机制，贯穿在几乎所有的流程中。弄明白这一块能够帮我们明白dubo源码
Dubbo是基于Java原生SPI机制思想的一个改进，所以，先从JAVA SPI机制开始了解什么时SPI以后再去学习Dubbo的SPI就比较简单
JAVASPI
SPI全称（service provider interface），是JDK内置的一种服务提供发现机制，目前市面上有很多框架都是用它来做服务的扩展发现，大家耳熟能详的如JDBC、日志框架都有用到；
简单来说，它是一种动态替换发现的机制。举个简单的例子，如果我们定义了一个规范，需要第三方厂商去实现，那么对于我们应用方来说，只需要集成对应厂商的插件，既可以完成对应规范的实现机制。 形成一种插拔式的扩展手段。
java原生spi实现
- 需要在classpath下创建一个目录，该目录命名必须是：
META-INF/services 在该目录下创建一个properties文件，该文件需要满足以下几个条件
- 文件名必须是扩展的接口的全路径名称
- 文件内部描述的是该扩展接口的所有实现类
- 文件的编码格式是UTF-8
- 通过java.util.ServiceLoader的加载机制来发现 首先定义api接口 接下来对应产商实现对应的接口，并且在
resouces的META-INF/services中创建对应的文件,并且通过properties规则配置实现类的全路径
以及对应调用方引入api接口，和对应产商的jar
并且在对应的resouces中引入接口，如果引入了多个产商的jar，那么会取到多个产商的东西
SPI的实际应用
SPI在很多地方有应用，大家可以看看最常用的java.sql.Driver驱动。JDK官方提供了java.sql.Driver这个驱动扩展点，但是你们并没有看到JDK中有对应的Driver实现。 那在哪里实现呢？
以连接Mysql为例，我们需要添加mysql-connector-java依赖。然后，你们可以在这个jar包中找到SPI的配置信息。如下图，所以java.sql.Driver由各个数据库厂商自行实现。这就是SPI的实际应用。当然除了这个意外，大家在spring的包中也可以看到相应的痕迹
SPI的缺点
- JDK标准的SPI会一次性加载实例化扩展点的所有实现，什么意思呢？就是如果你在META-INF/service下的文件里面加了N个实现类，那么JDK启动的时候都会一次性全部加载。那么如果有的扩展点实现初始化很耗时或者如果有些实现类并没有用到，那么会很浪费资源
- 如果扩展点加载失败，会导致调用方报错，而且这个错误很难定位到是这个原因
dubbo SPI规范
使用原生spi,如果路径下有多个实现都会加载进来，如果有一个加载失败，会比较麻烦
目录
Dubbo的SPI并非原生的SPI，Dubbo的规则是在
/META-INF/dubbo/META-INF/internal/META-INF/service
并且基于SPI接口去创建一个文件下面以需要实现的接口去创建一个文件，并且在文件中以properties规则一样配置实现类的全面以及分配实现的一个名称。
文件名称和接口名称保持一致，文件内容和SPI有差异，内容是key对应value
我们看一下dubbo-cluster模块的META-INF.dubbo.internal：
实现自己的扩展点
在resources目录下新建文件
META-INF/dubbo/com.alibaba.dubbo.rpc.Protocol文件，文件内容为
defineProtocol=com.gupaoedu.dubbo.protocol.DefineProtocol实现Protocol
import com.alibaba.dubbo.common.URL;import com.alibaba.dubbo.rpc.Exporter;import com.alibaba.dubbo.rpc.Invoker;import com.alibaba.dubbo.rpc.Protocol;import com.alibaba.dubbo.rpc.RpcException;public class DefineProtocol implements Protocol { @Override public int getDefaultPort() { return 8888; } @Override public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { return null; } @Override public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { return null; } @Override public void destroy() { }}调用
public class App { public static void main(String[] args) throws IOException, InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext ("dubbo-client.xml"); Protocol protocol=ExtensionLoader.getExtensionLoader(Protocol.class). getExtension("defineProtocol"); System.out.println(protocol.getDefaultPort()); System.in.read(); }}SPI源码分析
切入点
Protocol protocol=ExtensionLoader.getExtensionLoader(Protocol.class). getExtension("defineProtocol");dubbo的扩展点框架主要位于这个包下：
com.alibaba.dubbo.common.extension大概结构如下：
com.alibaba.dubbo.common.extension
|--factory|--AdaptiveExtensionFactory 自适应扩展点工厂 |--SpiExtensionFactory SPI扩展点工厂|--support
|--ActivateComparator|--Activate 自动激活加载扩展的注解
|--Adaptive 自适应扩展点的注解 |--ExtensionFactory 扩展点对象生成工厂接口|--ExtensionLoader 扩展点加载器，扩展点的查找，校验，加载等核心逻辑的实现类
|--SPI @SPI告诉当前应用其实一个扩展点例如Protocol 一定可以在对应的meta-inf/dubbo.internal 中看到dubbo-config-spring
|- extension
|--SpringExtensionFactory
其中最核心的类就是ExtensionLoader，几乎所有特性都在这个类中实现。
ExtensionLoader没有提供public的构造方法，但是提供了一个public static的getExtensionLoader，这个方法就是获取ExtensionLoader实例的工厂方法。其public成员方法中有三个比较重要的方法：
getActivateExtension ：根据条件获取当前扩展可自动激活的实现
getExtension ： 根据名称获取当前扩展的指定实现
getAdaptiveExtension : 获取当前扩展的自适应实现
Protocol源码
@SPI("dubbo")public interface Protocol { /** * 获取缺省端口，当用户没有配置端口时使用。 * * @return 缺省端口 */ int getDefaultPort(); /** * 暴露远程服务：<br> * 1. 协议在接收请求时，应记录请求来源方地址信息：RpcContext.getContext().setRemoteAddress();<br> * 2. export()必须是幂等的，也就是暴露同一个URL的Invoker两次，和暴露一次没有区别。<br> * 3. export()传入的Invoker由框架实现并传入，协议不需要关心。<br> * * @param <T> 服务的类型 * @param invoker 服务的执行体 * @return exporter 暴露服务的引用，用于取消暴露 * @throws RpcException 当暴露服务出错时抛出，比如端口已占用 */ @Adaptive <T> Exporter<T> export(Invoker<T> invoker) throws RpcException; /** * 引用远程服务：<br> * 1. 当用户调用refer()所返回的Invoker对象的invoke()方法时，协议需相应执行同URL远端export()传入的Invoker对象的invoke()方法。<br> * 2. refer()返回的Invoker由协议实现，协议通常需要在此Invoker中发送远程请求。<br> * 3. 当url中有设置check=false时，连接失败不能抛出异常，并内部自动恢复。<br> * * @param <T> 服务的类型 * @param type 服务的类型 * @param url 远程服务的URL地址 * @return invoker 服务的本地代理 * @throws RpcException 当连接服务提供方失败时抛出 */ @Adaptive <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException; /** * 释放协议：<br> * 1. 取消该协议所有已经暴露和引用的服务。<br> * 2. 释放协议所占用的所有资源，比如连接和端口。<br> * 3. 协议在释放后，依然能暴露和引用新的服务。<br> */ void destroy();}从上述Protocol的源码，可以看到两个注解@SPI("duo") 和 @Adaptive
@SPI :表示当前这个接口是一个扩展点，可以实现自己的扩展实现
@Adaptive 表示一个自适应扩展点，在方法级别上，会动态生成一个适配器类
getExtensionLoader
Protocol protocol=ExtensionLoader.getExtensionLoader(Protocol.class). getExtension("defineProtocol");@SuppressWarnings("unchecked")public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) { if (type == null) throw new IllegalArgumentException("Extension type == null"); if (!type.isInterface()) { throw new IllegalArgumentException("Extension type(" + type + ") is not interface!"); } if (!withExtensionAnnotation(type)) { throw new IllegalArgumentException("Extension type(" + type + ") is not extension, because WITHOUT @" + SPI.class.getSimpleName() + " Annotation!"); } ExtensionLoader<T> loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type); if (loader == null) { EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader<T>(type)); loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type); } return loader;}根据一个类型class获得一个ExtensionLoader，需要一个class类型的参数，这个参数必须是接口，而且此接口必须要@SPI注解注释，否则拒绝处理。检查通过之后首先会检查ExtensionLoader缓存中是否已经存在该扩展对应的ExtensionLoader,如果有则返回，否则创建一个新的ExtensionLoader负责加载此扩展实现，同时缓存起来。所以每一个扩展，dubbo中只会有一个对应的ExtensionLoader实例
接下来看下ExtensionLoader的私有构造函数：
private ExtensionLoader(Class<?> type) { this.type = type; objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension());}这里保存了对应的扩展类型，并且设置了一个额外的objectFactory属性，他是一个ExtensionFactory类型，ExtensionFactory主要用于加载扩展的实现：
ExtensionFactory主要用于加载扩展的实现：
@SPIpublic interface ExtensionFactory { /** * Get extension. * * @param type object type. * @param name object name. * @return object instance. */ <T> T getExtension(Class<T> type, String name);}ExtensionFactory有@SPI注解，说明当前这个接口是一个扩展点。从extension包的结构图可以看到。Dubbo内部提供了两个实现类：SpiExtensionFactory和AdaptiveExtensionFactory。不同的实现可以以不同的方式来完成扩展点实现的加载。
getAdaptiveExtension()
对应上述ExtensionLoader的getAdaptiveExtension()我们查看对应的getAdaptivesion()方法获得一个自适应的扩展点
如果是配置在类级别上，表示自定义适配器，如果是配置在方法上，表示需要动态生成适配器类
表示当前是自定义扩展点
默认的ExtensionFactory实现中，AdaptiveExtensionFactotry被@Adaptive注解注释，也就是它就是ExtensionFactory对应的自适应扩展实现(每个扩展点最多只能有一个自适应实现，如果所有实现中没有被@Adaptive注释的，那么dubbo会动态生成一个自适应实现类)，也就是说，所有对ExtensionFactory调用的地方，实际上调用的都是AdpativeExtensionFactory，那么我们看下他的实现代码：
@Adaptivepublic class AdaptiveExtensionFactory implements ExtensionFactory { private final List<ExtensionFactory> factories; public AdaptiveExtensionFactory() { ExtensionLoader<ExtensionFactory> loader = ExtensionLoader.getExtensionLoader(ExtensionFactory.class); List<ExtensionFactory> list = new ArrayList<ExtensionFactory>(); for (String name : loader.getSupportedExtensions()) { list.add(loader.getExtension(name)); } factories = Collections.unmodifiableList(list); } public <T> T getExtension(Class<T> type, String name) { for (ExtensionFactory factory : factories) { T extension = factory.getExtension(type, name); if (extension != null) { return extension; } } return null; }这段代码，其实就相当于一个代理入口，它会遍历当前系统中所有的ExtensionFactory实现来获取指定的扩展实现，获取到扩展实现，遍历完所有ExtensionFactory实现，调用ExtensionLoader的getSupportedExtensions方法来获取ExtensionFactory的所有实现
从前面ExtensionLoader的私有构造函数中可以看出，在选择ExtensionFactory的时候，并不是调用getExtension(name)来获取某个具体的实现类，而是调用getAdaptiveExtension来获取一个自适应的实现。那么首先我们就来分析一下getAdaptiveExtension这个方法的实现吧：
@SuppressWarnings("unchecked")public T getAdaptiveExtension() { Object instance = cachedAdaptiveInstance.get(); if (instance == null) { if(createAdaptiveInstanceError == null) { synchronized (cachedAdaptiveInstance) { instance = cachedAdaptiveInstance.get(); if (instance == null) { try { instance = createAdaptiveExtension(); cachedAdaptiveInstance.set(instance); } catch (Throwable t) { createAdaptiveInstanceError = t; throw new IllegalStateException("fail to create adaptive instance: " + t.toString(), t); } } } } else { throw new IllegalStateException("fail to create adaptive instance: " + createAdaptiveInstanceError.toString(), createAdaptiveInstanceError); } } return (T) instance;}首先检查缓存的adaptiveInstance是否存在，如果存在则直接使用，否则的话调用createAdaptiveExtension方法来创建新的adaptiveInstance并且缓存起来。也就是说对于某个扩展点，每次调用ExtensionLoader.getAdaptiveExtension获取到的都是同一个实例。
createAdaptiveExtension方法
private T createAdaptiveExtension() { try { return injectExtension((T) getAdaptiveExtensionClass().newInstance()); } catch (Exception e) { throw new IllegalStateException("Can not create adaptive extenstion " + type + ", cause: " + e.getMessage(), e); }} 在`createAdaptiveExtension`方法中，首先通过`getAdaptiveExtensionClass`方法获取到最终的自适应实现类型，然后实例化一个自适应扩展实现的实例，最后进行扩展点注入操作 private Class<?> getAdaptiveExtensionClass() { getExtensionClasses(); if (cachedAdaptiveClass != null) { return cachedAdaptiveClass; } return cachedAdaptiveClass = createAdaptiveExtensionClass();}上述代码中主要做了两件事情
getExtensionClasses()加载所有路径下的扩展点createAdaptiveExtensionClass()动态创建一个扩展点
cachedAdaptiveClass这里有个判断，用来判断当前Protocol这个扩展点是否存在一个自定义的适配器，如果有，则直接返回自定义适配器，否则，就动态创建，这个值是在getExtensionClasses中赋值的，这块代码我们稍后再看
getExtensionClasses()
private final Holder<Map<String, Class<?>>> cachedClasses = new Holder<Map<String,Class<?>>>();private Map<String, Class<?>> getExtensionClasses() { //com.alibaba.dubbo.rpc.Protocal=>[xx,xx] Map<String, Class<?>> classes = cachedClasses.get(); if (classes == null) { synchronized (cachedClasses) { classes = cachedClasses.get(); if (classes == null) { classes = loadExtensionClasses(); cachedClasses.set(classes); } } } return classes;}上述代码主要做了几件事
- 从cachedClasses中获得一个结果，这个结果实际上就是所有的扩展点类，key对应name,value对应class
- 通过双重检查锁判断
- 调用
loadExtensionClasses()，去加载扩展点的实现
loadExtensionClasses()
// 此方法已经getExtensionClasses方法同步过。private Map<String, Class<?>> loadExtensionClasses() { //type->Protocol.class final SPI defaultAnnotation = type.getAnnotation(SPI.class); if(defaultAnnotation != null) { String value = defaultAnnotation.value(); if(value != null && (value = value.trim()).length() > 0) { String[] names = NAME_SEPARATOR.split(value); if(names.length > 1) { throw new IllegalStateException("more than 1 default extension name on extension " + type.getName() + ": " + Arrays.toString(names)); } if(names.length == 1) cachedDefaultName = names[0]; } } Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>(); loadFile(extensionClasses, DUBBO_INTERNAL_DIRECTORY); loadFile(extensionClasses, DUBBO_DIRECTORY); loadFile(extensionClasses, SERVICES_DIRECTORY); return extensionClasses;}从代码里面可以看到，在loadExtensionClasses中首先会检测扩展点在@SPI注解中配置的默认扩展实现的名称，并将其赋值给cachedDefaultName属性进行缓存，后面想要获取该扩展点的默认实现名称就可以直接通过访问cachedDefaultName字段来完成，比如getDefaultExtensionName方法就是这么实现的。从这里的代码中又可以看到，具体的扩展实现类型，是通过调用loadFile方法来加载，分别从一下三个地方加载：
META-INF/dubbo/internal
META-INF/dubbo
META-INF/services/
主要逻辑：
- 获得当前扩展点的注解，也就是
Protocol.class这个类的注解@SPI - 判断这个注解不为空，则再次获取
@SPI中的value值 - 如果value有值，也就是
@SPI("dubbo")，则吧此dubbo的值赋值给cachedDefaultName，这就是为什么我们能够通过ExtensionLoader.getExtensionLoader(Protocol.class).getDefaultExtension()能够获取到DubboProtocol这个扩展点的原因 - 最后，通过loadFile去加载指定路径下的所有扩展点
loadFile()
private void loadFile(Map<String, Class<?>> extensionClasses, String dir) { String fileName = dir + type.getName(); try { Enumeration<java.net.URL> urls; ClassLoader classLoader = findClassLoader(); if (classLoader != null) { urls = classLoader.getResources(fileName); } else { urls = ClassLoader.getSystemResources(fileName); } if (urls != null) { while (urls.hasMoreElements()) { java.net.URL url = urls.nextElement(); try { BufferedReader reader = new BufferedReader(new InputStreamReader(url.openStream(), "utf-8")); try { String line = null; while ((line = reader.readLine()) != null) { final int ci = line.indexOf('#'); if (ci >= 0) line = line.substring(0, ci); line = line.trim(); if (line.length() > 0) { try { String name = null; int i = line.indexOf('='); if (i > 0) { name = line.substring(0, i).trim(); line = line.substring(i + 1).trim(); } if (line.length() > 0) { Class<?> clazz = Class.forName(line, true, classLoader); if (! type.isAssignableFrom(clazz)) { throw new IllegalStateException("Error when load extension class(interface: " + type + ", class line: " + clazz.getName() + "), class " + clazz.getName() + "is not subtype of interface."); } //如果在类级别上，表示自定义适配器 //如果是在方法上，表示需要动态生成适配器类 if (clazz.isAnnotationPresent(Adaptive.class)) { if(cachedAdaptiveClass == null) { cachedAdaptiveClass = clazz; } else if (! cachedAdaptiveClass.equals(clazz)) { throw new IllegalStateException("More than 1 adaptive class found: " + cachedAdaptiveClass.getClass().getName() + ", " + clazz.getClass().getName()); } } else { try { clazz.getConstructor(type); Set<Class<?>> wrappers = cachedWrapperClasses; if (wrappers == null) { cachedWrapperClasses = new ConcurrentHashSet<Class<?>>(); wrappers = cachedWrapperClasses; } wrappers.add(clazz); } catch (NoSuchMethodException e) { clazz.getConstructor(); if (name == null || name.length() == 0) { name = findAnnotationName(clazz); if (name == null || name.length() == 0) { if (clazz.getSimpleName().length() > type.getSimpleName().length() && clazz.getSimpleName().endsWith(type.getSimpleName())) { name = clazz.getSimpleName().substring(0, clazz.getSimpleName().length() - type.getSimpleName().length()).toLowerCase(); } else { throw new IllegalStateException("No such extension name for the class " + clazz.getName() + " in the config " + url); } } } String[] names = NAME_SEPARATOR.split(name); if (names != null && names.length > 0) { Activate activate = clazz.getAnnotation(Activate.class); if (activate != null) { cachedActivates.put(names[0], activate); } for (String n : names) { if (! cachedNames.containsKey(clazz)) { cachedNames.put(clazz, n); } Class<?> c = extensionClasses.get(n); if (c == null) { extensionClasses.put(n, clazz); } else if (c != clazz) { throw new IllegalStateException("Duplicate extension " + type.getName() + " name " + n + " on " + c.getName() + " and " + clazz.getName()); } } } } } } } catch (Throwable t) { IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: " + type + ", class line: " + line + ") in " + url + ", cause: " + t.getMessage(), t); exceptions.put(line, e); } } } // end of while read lines } finally { reader.close(); } } catch (Throwable t) { logger.error("Exception when load extension class(interface: " + type + ", class file: " + url + ") in " + url, t); } } // end of while urls } } catch (Throwable t) { logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName + ").", t); }}解析指定路径下的文件，获取对应的扩展点，通过反射的方式进行实例化之后，put到extensionClasses这个Map集合中
调用loadFile方法，代码比较长，主要做了几个事情，有几个变量会赋值
cachedAdaptiveClass : 当前Extension类型对应的AdaptiveExtension类型(只能一个)
cachedWrapperClasses : 当前Extension类型对应的所有Wrapper实现类型(无顺序)
cachedActivates : 当前Extension实现自动激活实现缓存(map,无序)
cachedNames : 扩展点实现类对应的名称(如配置多个名称则值为第一个)
当loadExtensionClasses方法执行完成之后，还有以下变量被赋值：
cachedDefaultName : 当前扩展点的默认实现名称
当getExtensionClasses方法执行完成之后，除了上述变量被赋值之外，还有以下变量被赋值：
cachedClasses : 扩展点实现名称对应的实现类(一个实现类可能有多个名称)
其实也就是说，在调用了getExtensionClasses方法之后，当前扩展点对应的实现类的一些信息就已经加载进来了并且被缓存了。后面的许多操作都可以直接通过这些缓存数据来进行处理了。
createAdaptiveExtensionClass()
private Class<?> createAdaptiveExtensionClass() { //生成字节码代码 String code = createAdaptiveExtensionClassCode(); //获得类加载器 ClassLoader classLoader = findClassLoader(); com.alibaba.dubbo.common.compiler.Compiler compiler = ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.common.compiler.Compiler.class).getAdaptiveExtension(); //动态编译字节码 return compiler.compile(code, classLoader);}动态生成适配器代码，以及动态编译
- createAdaptiveExtensionClassCode ,动态创建一个字节码文件，返回code这个字符串
- 通过
compilier.compile进行编译（默认情况下使用的是javassist） - 通过classLoader载入到jvm中
上生成的code和类是
import com.alibaba.dubbo.common.extension.ExtensionLoader;public class Protocol$Adaptive implements com.alibaba.dubbo.rpc.Protocol { public void destroy() { throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!"); } public int getDefaultPort() { throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!"); } public com.alibaba.dubbo.rpc.Invoker refer(java.lang.Class arg0, com.alibaba.dubbo.common.URL arg1) throws com.alibaba.dubbo.rpc.RpcException { if (arg1 == null) throw new IllegalArgumentException("url == null"); com.alibaba.dubbo.common.URL url = arg1; String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol()); if (extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])"); com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName); return extension.refer(arg0, arg1); } public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null"); if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null"); com.alibaba.dubbo.common.URL url = arg0.getUrl(); String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol()); if (extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])"); com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName); return extension.export(arg0); }}动态生成类的主要功能
- 从url或者拓展接口获取扩展接口实现类的名称
- 根据名称，获取实现了
ExtensionLoader.getExtensionLoader(扩展接口类).getExtension(扩展接口实现类名称)，然后调用类的方法
refer是一个引用，得到一个url的参数，通过参数判断是走哪个协议发布服务
回到createAdaptiveExtension方法，他调用了getExtesionClasses方法加载扩展点实现信息完成之后，就可以直接通过判断cachedAdaptiveClass缓存字段是否被赋值盘确定当前扩展点是否有默认的AdaptiveExtension实现。如果没有，那么就调用createAdaptiveExtensionClass方法来动态生成一个。在dubbo的扩展点框架中大量的使用了缓存技术。
创建自适应扩展点实现类型和实例化就已经完成了，下面就来看下扩展点自动注入的实现
injectExtension
@Adaptivepublic class AdaptiveExtensionFactory implements ExtensionFactory { private final List<ExtensionFactory> factories; public AdaptiveExtensionFactory() { ExtensionLoader<ExtensionFactory> loader = ExtensionLoader.getExtensionLoader(ExtensionFactory.class); List<ExtensionFactory> list = new ArrayList<ExtensionFactory>(); for (String name : loader.getSupportedExtensions()) { list.add(loader.getExtension(name)); } factories = Collections.unmodifiableList(list); } public <T> T getExtension(Class<T> type, String name) { for (ExtensionFactory factory : factories) { T extension = factory.getExtension(type, name); if (extension != null) { return extension; } } return null; }}这里可以看到，扩展点自动注入的一句就是根据setter方法对应的参数类型和property名称从ExtensionFactory中查询，如果有返回扩展点实例，那么就进行注入操作。到这里getAdaptiveExtension方法就分析完毕了。
需要明白一点dubbo的内部传参基本上都是基于Url来实现的，也就是说Dubbo是基于URL驱动的技术
所以，适配器类的目的是在运行期获取扩展的真正实现来调用，解耦接口和实现，这样的话要不我们自己实现适配器类，要不dubbo帮我们生成，而这些都是通过Adpative来实现。
到目前为止，我们的AdaptiveExtension的主线走完了，可以简单整理一下他们的调用关系如下
总结
在整个过程中，最重要的两个方法getExtensionClasses和createAdaptiveExtensionClass
getExtensionClasses
这个方法主要是读取META-INF/services 、META-INF/dubbo、META-INF/internal目录下的文件内容
分析每一行，如果发现其中有哪个类的annotation是@Adaptive，就找到对应的AdaptiveClass。如果没有的话，就动态创建一个
createAdaptiveExtensionClass
该方法是在getExtensionClasses方法找不到AdaptiveClass的情况下被调用，该方法主要是通过字节码的方式在内存中新生成一个类，它具有AdaptiveClass的功能，Protocol就是通过这种方式获得AdaptiveClass类的。
源码分析
加载顺序
NamespaceHandler: 注册BeanDefinitionParser， 利用它来解析
BeanDefinitionParser： 解析配置文件的元素
spring会默认加载jar包下/META-INF/spring.handlers 找到对应的NamespaceHandler
initializingBean
当spring容器初始化完以后，会调用afterPropertiesSet方法
DisposableBean
bean被销毁的时候调用destory方法
ApplicationContextAware
容器初始化完成之后会主动注入applicationContext
ApplicationListener
事件监听
BeanNameAware
对象初始化完之后会获取bean的本身属性
delay能够控制延迟发布
源码解读要点
首先我们要关注的是服务的发布和服务的消费这两个主要的流程，那么就可以基于这个点去找到源码分析的突破口。那么自然而然我们就可以想到spring的配置
Dubbo的接入实现
Dubbo中spring扩展就是使用spring的自定义类型，所以同样也有NamespaceHandler、BeanDefinitionParser。而NamespaceHandler是DubboNamespaceHandler
public class DubboNamespaceHandler extends NamespaceHandlerSupport { static { Version.checkDuplicate(DubboNamespaceHandler.class); } public void init() { registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true)); registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true)); registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true)); registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true)); registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true)); registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true)); registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true)); registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true)); registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false)); registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true)); }} BeanDefinitionParser全部都使用了DubboBeanDefinitionParser，如果我们向看<dubbo:service>的配置，就直接看DubboBeanDefinitionParser中
这个里面主要做了一件事，把不同的配置分别转化成spring容器中的bean对象
application对应ApplicationConfig
registry对应RegistryConfig
monitor对应MonitorConfig
provider对应ProviderConfig
consumer对应ConsumerConfig
为了在spring启动的时候，也相应的启动provider发布服务注册服务的过程，而同时为了让客户端在启动的时候自动订阅发现服务，加入了两个bean
ServiceBean、ReferenceBean。
分别继承了ServiceConfig和ReferenceConfig
同时还分别实现了InitializingBean、DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware
InitializingBean接口为bean提供了初始化方法的方式，它只包括afterPropertiesSet方法，凡是继承该接口的类，在初始化bean的时候会执行该方法。
DisposableBean bean被销毁的时候，spring容器会自动执行destory方法，比如释放资源
ApplicationContextAware 实现了这个接口的bean，当spring容器初始化的时候，会自动的将ApplicationContext注入进来
ApplicationListener ApplicationEvent事件监听，spring容器启动后会发一个事件通知
BeanNameAware 获得自身初始化时，本身的bean的id属性
那么基本的实现思路可以整理出来了
- 利用spring的解析收集xml中的配置信息，然后把这些配置信息存储到serviceConfig中
- 调用ServiceConfig的export方法来进行服务的发布和注册
服务的发布
ServiceBean
serviceBean是服务发布的切入点，通过afterPropertiesSet方法，调用export()方法进行发布。
export为父类ServiceConfig中的方法，所以跳转到SeviceConfig类中的export方法
public synchronized void export() { if (provider != null) { if (export == null) { export = provider.getExport(); } if (delay == null) { delay = provider.getDelay(); } } if (export != null && ! export.booleanValue()) { return; } if (delay != null && delay > 0) { Thread thread = new Thread(new Runnable() { public void run() { try { Thread.sleep(delay); } catch (Throwable e) { } doExport(); } }); thread.setDaemon(true); thread.setName("DelayExportServiceThread"); thread.start(); } else { doExport(); }}我们发现，delay的作用就是延迟暴露，而延迟的方式也很直截了当，Thread.sleep(delay)
- export是synchronized修饰的方法。也就是说暴露的过程是原子操作，正常情况下不会出现锁竞争的问题，毕竟初始化过程大多数情况下都是单一线程操作，这里联想到了spring的初始化流程，也进行了加锁操作，这里也给我们平时设计一个不错的启示：初始化流程的性能调优优先级应该放的比较低，但是安全的优先级应该放的比较高！
- 继续看
doExport()方法。同样是一堆初始化代码
export的过程
继续看doExport()，最终会调用到doExportUrls()中：
private void doExportUrls() { List<URL> registryURLs = loadRegistries(true);//是不是获得注册中心的配置 for (ProtocolConfig protocolConfig : protocols) { //是不是支持多协议发布 doExportUrlsFor1Protocol(protocolConfig, registryURLs); }}对应上述protocols长成这样<dubbo:protocol name="dubbo" port="20880" id ="dubbo"/>protocols也是根据配置装配出来的，接下来进入对应的duExportUrlsFor1Protocol方法查看对应s具体实现
最终实现逻辑
//配置为none不暴露if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { //配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务) if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) { exportLocal(url); } //如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露本地服务) if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){ if (logger.isInfoEnabled()) { logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url); } if (registryURLs != null && registryURLs.size() > 0 && url.getParameter("register", true)) { for (URL registryURL : registryURLs) { url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic")); URL monitorUrl = loadMonitor(registryURL); if (monitorUrl != null) { url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString()); } if (logger.isInfoEnabled()) { logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL); } Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); Exporter<?> exporter = protocol.export(invoker); exporters.add(exporter); } } else { Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url); Exporter<?> exporter = protocol.export(invoker); exporters.add(exporter); } }}从上述代码doExportUrlsFor1Protocol方法，先创建两个URL，分别如下
dubbo://192.168.xx.xx:20880/com.zzjson.IHello
registry://192.168.xx
其对应的url就是services的providers的信息
在上面这段代码中可以看到Dubbo的比较核心的抽象：Invoker， Invoker是一个代理类，从ProxyFactory中生成。这个地方可以做一个小结
- Invoker -执行具体的远程调用
- Protocol – 服务地址的发布和订阅
- Exporter – 暴露服务或取消暴露
Protocol$Adaptive
public class Protocol$Adaptive implements com.alibaba.dubbo.rpc.Protocol { public void destroy() { throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!"); } public int getDefaultPort() { throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!"); } public com.alibaba.dubbo.rpc.Invoker refer(java.lang.Class arg0, com.alibaba.dubbo.common.URL arg1) throws com.alibaba.dubbo.rpc.RpcException { if (arg1 == null) throw new IllegalArgumentException("url == null"); com.alibaba.dubbo.common.URL url = arg1; String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol()); if (extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])"); com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName); return extension.refer(arg0, arg1); } public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null"); if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null"); com.alibaba.dubbo.common.URL url = arg0.getUrl(); String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol()); if (extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])"); com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName); return extension.export(arg0); }}上述代码做两件事
- 从url中获得
protocol的协议地址，如果protocol为空，表示通过dubbo协议发布服务，否则根据配置的谢意类型来发布服务 - 调用
ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(extName)
getExtension()
public T getExtension(String name) { if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null"); if ("true".equals(name)) { return getDefaultExtension(); } Holder<Object> holder = cachedInstances.get(name); if (holder == null) { cachedInstances.putIfAbsent(name, new Holder<Object>()); holder = cachedInstances.get(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;}createExtension()
@SuppressWarnings("unchecked")private T createExtension(String name) { 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, (T) clazz.newInstance()); instance = (T) EXTENSION_INSTANCES.get(clazz); } //对获取的实例进行依赖注入 injectExtension(instance); //在loadFile中进行赋值的 Set<Class<?>> wrapperClasses = cachedWrapperClasses; if (wrapperClasses != null && wrapperClasses.size() > 0) { for (Class<?> wrapperClass : wrapperClasses) { //对实例进行包装，分别调用带Protocol参数的构造函数创建实例，然后进行依赖注入 instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance)); } } return instance; } catch (Throwable t) { throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type + ") could not be instantiated: " + t.getMessage(), t); }}主要做如下四个事情
- 根据name获取对应的class
- 根据calss创建一个实例
- 对获取的实例进行依赖注入
- 对实例进行包装，分别调用带
Protocol参数的构造函数创建实例，然后进行依赖注入 - 在
dubbo-rpc-api的resources路径下，找到com.alibaba.dubbo.rpc.Protocol文件中存在filter/listener - 遍历
cachedWrapperClass对DubboProtocol进行包装，会通过ProtocolFilterWrapper,ProtocolListenerWrapper包装
private Map<String, Class<?>> getExtensionClasses() { Map<String, Class<?>> classes = cachedClasses.get(); if (classes == null) { synchronized (cachedClasses) { classes = cachedClasses.get(); if (classes == null) { classes = loadExtensionClasses(); cachedClasses.set(classes); } } } return classes;}总结
在ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(extName);这段代码中，当extName为registry的时候，我们不需要再次去阅读这块代码了，直接可以在扩展点中找到相应的实现扩展类，/dubbo-registry-api/src/main/resources/META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Protocol配置如下
registry=com.alibaba.dubbo.registry.integration.RegistryProtocol所以我们定位到RegistryProtocolRegistryProtocol这个类中的export方法
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { //export invoker final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker); //registry provider final Registry registry = getRegistry(originInvoker); final URL registedProviderUrl = getRegistedProviderUrl(originInvoker); registry.register(registedProviderUrl); // 订阅override数据 // FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。 final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl); final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl); overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener); //保证每次export都返回一个新的exporter实例 return new Exporter<T>() { public Invoker<T> getInvoker() { return exporter.getInvoker(); } public void unexport() { try { exporter.unexport(); } catch (Throwable t) { logger.warn(t.getMessage(), t); } try { registry.unregister(registedProviderUrl); } catch (Throwable t) { logger.warn(t.getMessage(), t); } try { overrideListeners.remove(overrideSubscribeUrl); registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener); } catch (Throwable t) { logger.warn(t.getMessage(), t); } } };}doLoalExport()
上述doLoaclExport源码为：
private Protocol protocol;public void setProtocol(Protocol protocol) { this.protocol = protocol;}@SuppressWarnings("unchecked")private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker){ String key = getCacheKey(originInvoker); ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key); if (exporter == null) { synchronized (bounds) { exporter = (ExporterChangeableWrapper<T>) bounds.get(key); if (exporter == null) { final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker)); exporter = new ExporterChangeableWrapper<T>((Exporter<T>)protocol.export(invokerDelegete), originInvoker); bounds.put(key, exporter); } } } return (ExporterChangeableWrapper<T>) exporter;}上述代码Protocol代码是怎么复制的呢，是在injectExtension方法中对已经加载的扩展点的属性进行依赖注入了
protocol发布服务
因此我们知道protocol是一个自适应扩展点，Protocol$Adaptive，然后调用这个自适应扩展点中的export方法，这个时候传入的协议地址应该是
dubbo://127.0.0.1/xxx因此在Protocol$Adaptive.export方法中，ExtensionLoader.getExtension(Protocol.class).getExtension就是基于DubboProtocol协议发布服务了么？当然不是，此处获取的不是一个单纯的DubboProtocol扩展点，而是会通过Wrapper对Protocl进行装饰，装饰器分别为ProtocolFilterWrapper或者是ProtoclListenerWrapper，至于为什么MockProtocol为什么不在装饰器里面呢？我们可以想到，在ExtensionLoader.loadFile这段代码的时候，装饰器必须要有一个Protocol的构造方法，如下
public ProtocolFilterWrapper(Protocol protocol){ if (protocol == null) { throw new IllegalArgumentException("protocol == null"); } this.protocol = protocol;}至此，我们可以知道Protocol$Adaptive中的export方法会调用ProtocolFilterWrapper以及ProtocolListenerWrapper类的方法
ProtocolFilterWrapper
public class ProtocolFilterWrapper implements Protocol { private final Protocol protocol; public ProtocolFilterWrapper(Protocol protocol) { if (protocol == null) { throw new IllegalArgumentException("protocol == null"); } this.protocol = protocol; }//此方法读取所有的filter类，利用这些类封装invoker 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 { return filter.invoke(next, invocation); } @Override public void destroy() { invoker.destroy(); } @Override public String toString() { return invoker.toString(); } }; } } return last; } @Override public int getDefaultPort() { return protocol.getDefaultPort(); } @Override public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER)); } @Override public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } return buildInvokerChain(protocol.refer(type, url), Constants.REFERENCE_FILTER_KEY, Constants.CONSUMER); } @Override public void destroy() { protocol.destroy(); }}ProtocolFilterWrapper这个类非常重要
- 他有一个参数为
Protocol protocol构造参数 - 它实现了
Protocol接口 - 使用了责任链模式，对
export和refer函数进行了封装
我们查看如下文件dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Filter
echo=com.alibaba.dubbo.rpc.filter.EchoFiltergeneric=com.alibaba.dubbo.rpc.filter.GenericFiltergenericimpl=com.alibaba.dubbo.rpc.filter.GenericImplFiltertoken=com.alibaba.dubbo.rpc.filter.TokenFilteraccesslog=com.alibaba.dubbo.rpc.filter.AccessLogFilteractivelimit=com.alibaba.dubbo.rpc.filter.ActiveLimitFilterclassloader=com.alibaba.dubbo.rpc.filter.ClassLoaderFiltercontext=com.alibaba.dubbo.rpc.filter.ContextFilterconsumercontext=com.alibaba.dubbo.rpc.filter.ConsumerContextFilterexception=com.alibaba.dubbo.rpc.filter.ExceptionFilterexecutelimit=com.alibaba.dubbo.rpc.filter.ExecuteLimitFilterdeprecated=com.alibaba.dubbo.rpc.filter.DeprecatedFiltercompatible=com.alibaba.dubbo.rpc.filter.CompatibleFiltertimeout=com.alibaba.dubbo.rpc.filter.TimeoutFilter可以看到，invoker通过如下的Filter组装成一个责任链
其中涉及到很多功能，包括权限验证，异常，超时等等，当然可以预计计算调用时间等等应该也是在这其中的某个类实现的，这里我们可以看到export和refer过程都会被filter过滤
ProtocolListenerWrapper
在这里我们可以看到export和refer分别对应了不同的Wrapper;export对应的ListenerExporterWrapper这块暂不去分析，因为此地方并没有提供实现类
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 int getDefaultPort() { return protocol.getDefaultPort(); } public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } return new ListenerExporterWrapper<T>(protocol.export(invoker), Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class) .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY))); } public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } return new ListenerInvokerWrapper<T>(protocol.refer(type, url), Collections.unmodifiableList( ExtensionLoader.getExtensionLoader(InvokerListener.class) .getActivateExtension(url, Constants.INVOKER_LISTENER_KEY))); } public void destroy() { protocol.destroy(); }}我们看一下dubboProtocol的export方法：openServer(url）
DuoProtocol.export()
通过上述代码分析完以后，我们能够定位到DubboProtocol.export()方法
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { URL url = invoker.getUrl(); // export service. String key = serviceKey(url); DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap); exporterMap.put(key, exporter); //export an stub service for dispaching event Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT); Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false); if (isStubSupportEvent && !isCallbackservice){ String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY); if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){ if (logger.isWarnEnabled()){ logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) + "], has set stubproxy support event ,but no stub methods founded.")); } } else { stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods); } }//暴露服务 openServer(url); return exporter;}接着调用openServer
private void openServer(URL url) { // find server. String key = url.getAddress();//116.62.221.6:20880 //client 也可以暴露一个只有server可以调用的服务。 boolean isServer = url.getParameter(Constants.IS_SERVER_KEY,true); if (isServer) { ExchangeServer server = serverMap.get(key); if (server == null) {//没有的话就创建服务 serverMap.put(key, createServer(url)); } else { //server支持reset,配合override功能使用 server.reset(url); } }}createServer()
继续看其中的createServer方法：
private ExchangeServer createServer(URL url) { //默认开启server关闭时发送readonly事件 url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()); //默认开启heartbeat url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT)); String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER); if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) throw new RpcException("Unsupported server type: " + str + ", url: " + url); url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME); ExchangeServer server; try { server = Exchangers.bind(url, requestHandler); } catch (RemotingException e) { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } str = url.getParameter(Constants.CLIENT_KEY); if (str != null && str.length() > 0) { Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(); if (!supportedTypes.contains(str)) { throw new RpcException("Unsupported client type: " + str); } } return server;}创建服务，然后开启心跳监测，默认使用netty 。组装url
Exchanger.bind()
发现ExchangeServer是通过Exchangers创建的，直接看Exchanger.bind方法
public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { if (url == null) { throw new IllegalArgumentException("url == null"); } if (handler == null) { throw new IllegalArgumentException("handler == null"); } url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange"); return getExchanger(url).bind(url, handler);}getExchanger方法实际上调用的是ExtensionLoader的相关方法，这里的ExtensionLoader是dubbo插件化的核心，我们会在后面的插件化讲解中详细讲解，这里我们只需要知道Exchanger的默认实现只有一个：HeaderExchanger。上面一段代码最终调用的是：
public static Exchanger getExchanger(URL url) { String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER); return getExchanger(type);}public static Exchanger getExchanger(String type) { return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);}public class HeaderExchanger implements Exchanger { public static final String NAME = "header"; public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); } public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); }}可以看到Server与Client实例均是在这里创建的，HeaderExchangeServer需要一个Server类型的参数，来自Transporters.bind()：
Transporters.bind()
public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException { if (url == null) { throw new IllegalArgumentException("url == null"); } if (handlers == null || handlers.length == 0) { throw new IllegalArgumentException("handlers == null"); } ChannelHandler handler; if (handlers.length == 1) { handler = handlers[0]; } else { handler = new ChannelHandlerDispatcher(handlers); } return getTransporter().bind(url, handler);} getTransporter()获取的实例来源于配置，默认返回一个NettyTransporter：
NettyTransport.bind()
通过NettyTranport创建基于Netty的server服务
public class NettyTransporter implements Transporter { public static final String NAME = "netty"; public Server bind(URL url, ChannelHandler listener) throws RemotingException { return new NettyServer(url, listener); } public Client connect(URL url, ChannelHandler listener) throws RemotingException { return new NettyClient(url, listener); }}在调用HeaderExchanger.bind()方法的时候，是先new一个HeaderExchangeServer这个server是干嘛呢，是对当前这个链接去建立心跳机制
public class HeaderExchangeServer implements ExchangeServer { protected final Logger logger = LoggerFactory.getLogger(getClass()); private final ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1,new NamedThreadFactory("dubbo-remoting-server-heartbeat", true); // 心跳定时器 private ScheduledFuture<?> heatbeatTimer; // 心跳超时，毫秒。缺省0，不会执行心跳。 private int heartbeat; private int heartbeatTimeout; private final Server server; private volatile boolean closed = false; public HeaderExchangeServer(Server server) { if (server == null) { throw new IllegalArgumentException("server == null"); } this.server = server; this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0); this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3); if (heartbeatTimeout < heartbeat * 2) { throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2"); } //心跳 startHeatbeatTimer(); } private void startHeatbeatTimer() { //关闭心跳定时 stopHeartbeatTimer(); if (heartbeat > 0) { //每隔heartbeat时间执行一次 heatbeatTimer = scheduled.scheduleWithFixedDelay( new HeartBeatTask( new HeartBeatTask.ChannelProvider() { //获取channels public Collection<Channel> getChannels() { return Collections.unmodifiableCollection( HeaderExchangeServer.this.getChannels() ); } }, heartbeat, heartbeatTimeout), heartbeat, heartbeat,TimeUnit.MILLISECONDS); } } //关闭心跳定时 private void stopHeartbeatTimer() { try { ScheduledFuture<?> timer = heatbeatTimer; if (timer != null && ! timer.isCancelled()) { timer.cancel(true); } } catch (Throwable t) { logger.warn(t.getMessage(), t); } finally { heatbeatTimer =null; } }}心跳线程HeartBeatTask在超时时间之内发送数据，在超时时间之外，是客户端的话，重连；是服务端，那么关闭服务端发布。
服务的注册
前面我们知道，基于Spring的这个解析入口，到发布服务的过程，接着基于DubboProtocol去发布，最终调用Netty的api创建了一个NettyServer。
那么继续沿着RegistryProtocol.export这个方法，来看看注册服务的代码
RegistryProtocol.export()
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { //export invoker final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker); //registry provider final Registry registry = getRegistry(originInvoker); final URL registedProviderUrl = getRegistedProviderUrl(originInvoker); registry.register(registedProviderUrl); // 订阅override数据 // FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。 final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl); final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl); overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener); //保证每次export都返回一个新的exporter实例 return new Exporter<T>() { public Invoker<T> getInvoker() { return exporter.getInvoker(); } public void unexport() { try { exporter.unexport(); } catch (Throwable t) { logger.warn(t.getMessage(), t); } try { registry.unregister(registedProviderUrl); } catch (Throwable t) { logger.warn(t.getMessage(), t); } try { overrideListeners.remove(overrideSubscribeUrl); registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener); } catch (Throwable t) { logger.warn(t.getMessage(), t); } } };}getRegistry()
private RegistryFactory registryFactory;public void setRegistryFactory(RegistryFactory registryFactory) { this.registryFactory = registryFactory;}private Registry getRegistry(final Invoker<?> originInvoker){ URL registryUrl = originInvoker.getUrl();//获得registry://192.168.xx.xx:2181的协议地址 if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) { //得到zk的谢意地址 String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY); //registryUrl就会变成了zookeeper://192.168.xx.xx registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY); } //registryFactory是什么 return registryFactory.getRegistry(registryUrl);}上述代码很明显了，通过分析，其实就是把registry的协议头改成服务提供者配置的协议地址，就是我们配置
的
<dubbo:registry address="zookeeper://192.168.xx.xx.2181"/>然后registryFactory.getRegistry的目的，就是通过协议地址匹配到对应的注册中心。那registryFactory是一个什么样的对象呢，从上述我们可以才，其是一个扩展点，并且我们能够注意到这里面的一个方法上有一个@Adaptive的注解，说明了其实一个自适应扩展点，按照我们之前看过的代码，自适应扩展点加在方法层面上，表示会动态生成一个自适应的适配器，所以这个自适应适配器应该是RegistryFactory$Adaptive
@SPI("dubbo")public interface RegistryFactory { /** * 连接注册中心. * * 连接注册中心需处理契约：<br> * 1. 当设置check=false时表示不检查连接，否则在连接不上时抛出异常。<br> * 2. 支持URL上的username:password权限认证。<br> * 3. 支持backup=10.20.153.10备选注册中心集群地址。<br> * 4. 支持file=registry.cache本地磁盘文件缓存。<br> * 5. 支持timeout=1000请求超时设置。<br> * 6. 支持session=60000会话超时或过期设置。<br> * * @param url 注册中心地址，不允许为空 * @return 注册中心引用，总不返回空 */ @Adaptive({"protocol"}) Registry getRegistry(URL url);}RegistryFactory$Adaptive
public class RegistryFactory$Adaptive implements com.alibaba.dubbo.registry.RegistryFactory { public com.alibaba.dubbo.registry.Registry getRegistry(com.alibaba.dubbo.common.URL arg0) { if (arg0 == null) throw new IllegalArgumentException("url == null"); com.alibaba.dubbo.common.URL url = arg0; String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol()); if (extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.registry.RegistryFactory) " +"name from url(" + url.toString() + ") usekeys([protocol])")"; com.alibaba.dubbo.registry.RegistryFactory extension = (com.alibaba.dubbo.registry.RegistryFactory) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.registry.Reg istryFactory.class).getExtension(extName); return extension.getRegistry(arg0); }}我们拿到这个动态生成的自适应扩展点，看看这段代码中的实现
- 从url中拿到协议头信息，这个时候的协议头是
zookeeper:// - 通过
ExtensionLoader.getExtensionLoader(RegistryFactory.class).getExtension("zookeeper")去获得一个指定的扩展点，而这个扩展点的配置在
/dubbo-registry/dubbo-registry-zookeeper/src/main/resources/META-INF/dubbo/internal/com.alibaba.dubbo.registry.RegistryFactory内容为
zookeeper=com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistryFactoryZookeeperRegistryFactory
public class ZookeeperRegistryFactory extends AbstractRegistryFactory { private ZookeeperTransporter zookeeperTransporter; public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) { this.zookeeperTransporter = zookeeperTransporter; } public Registry createRegistry(URL url) { return new ZookeeperRegistry(url, zookeeperTransporter); }}此方法中并没有getRegistry方法，而是在父类AbstractRegistryFactory
public Registry getRegistry(URL url) { url = url.setPath(RegistryService.class.getName()) .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName()) .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY); String key = url.toServiceString(); // 锁定注册中心获取过程，保证注册中心单一实例 LOCK.lock(); try { Registry registry = REGISTRIES.get(key); if (registry != null) { return registry; } registry = createRegistry(url); if (registry == null) { throw new IllegalStateException("Can not create registry " + url); } REGISTRIES.put(key, registry); return registry; } finally { // 释放锁 LOCK.unlock(); }}上述方法
- 从缓存
REGISTRIES中，根据key获得对应的Registry - 如果不存在，则创建
createRegistry
public Registry createRegistry(URL url) { return new ZookeeperRegistry(url, zookeeperTransporter); }public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) { super(url); if (url.isAnyHost()) { throw new IllegalStateException("registry address == null"); } String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT); if (! group.startsWith(Constants.PATH_SEPARATOR)) { group = Constants.PATH_SEPARATOR + group; } this.root = group; zkClient = zookeeperTransporter.connect(url); zkClient.addStateListener(new StateListener() { public void stateChanged(int state) { if (state == RECONNECTED) { try { recover(); } catch (Exception e) { logger.error(e.getMessage(), e); } } } });}代码分析到这，我们对于getRegistry得出结论根据当前注册中心的配置信息，获得一个匹配的注册中心，也就是ZookeeperRegistry registry.register(registedProviderUrl)
继续往下看会调用对应的registry.register去把dubbo://的谢意地址注册到zookeeper上，这个方法会调用FailbackRegistry类中的register,因为其父类FailbackRegistry中存在register方法，而这个类重写了此方法，所以我们可以直接定位到FailbackRegistry这个类中的register方法中
FailbackRegistry.register
@Overridepublic void register(URL url) { super.register(url); failedRegistered.remove(url); failedUnregistered.remove(url); try { // 向服务器端发送注册请求 doRegister(url); } catch (Exception e) { Throwable t = e; // 如果开启了启动时检测，则直接抛出异常 boolean check = getUrl().getParameter(Constants.CHECK_KEY, true) && url.getParameter(Constants.CHECK_KEY, true) && ! Constants.CONSUMER_PROTOCOL.equals(url.getProtocol()); boolean skipFailback = t instanceof SkipFailbackWrapperException; if (check || skipFailback) { if(skipFailback) { t = t.getCause(); } throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t); } else { logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t); } // 将失败的注册请求记录到失败列表，定时重试 failedRegistered.add(url); }}从名字上来看，是一个失败重试机制，调用父类的register方法把当前url添加到缓存集合中，调用子类的doRegister方法
protected void doRegister(URL url) { try { zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true)); } catch (Throwable e) { throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); }}可以看到，调用了zkclient.create在zookeeper中创建节点
最后RegistryProtocol.export这个方法之后的代码不再分析了，就是去服务提供端注册一个zookeeper监听，当监听发生变化的时候，服务端做相应的处理。