Dubbo实现原理分析-SPI&自适应扩展实现

Duboo基本概念Dubbo整体架构：evernotecid://D5D92FBE-C671-4B13-BAC8-4D01D3D20F5B/appyinxiangcom/8739769/ENNote/p68?hash=f07431ff8daafff6906882197d395a2aDubbo SPISPI 全称为 Service Provider Interface，一种服务提供发现机制。可以实现通过配置手段加载相应的实现类。JDK里提供了一种SPI实现,Dubbo并没有使用jdk的spi，而是自己实现了一个Dubbo SPI 服务发现机制。首先看一下JDK SPI实现方式定义一个接口类Travelpublic interface Travel { public void travel();}Travel有两个实现类public class CarTravel implements Travel { @Override public void travel() { System.out.println(“travel by car”); }}public class PlaneTravel implements Travel{ @Override public void travel() { System.out.println(“travel by plane”); }}接下来需要在 META-INF/services 文件夹下建立接口名称对应的文件，META-INF/services/com.demo.spi.jdk.Travel文件内容为接口实现类的名称com.demo.spi.jdk.CarTravelcom.demo.spi.jdk.PlaneTravel测试使用public class JdkSpiTest { public static void main(String[] args) { ServiceLoader<Travel> sServiceLoader = ServiceLoader.load(Travel.class); sServiceLoader.forEach(Travel::travel); }}测试结果travel by cartravel by planeJAVA SPI vs Dubbo SPI首先加载路径不同，java spi路径是 META-INF/services ,dubbo 是META-INF/services/，META-INF/services/internal，META-INF/dubbo/加载方式不同，java spi是全量加载，dubbo是按需加载接口实现类dubbo spi除了加载实现类外还增加了 IOC 和 AOP 特性下面我们看一下dubbo spi的实现首先接口需要有@SPI注解@Documented@Retention(RetentionPolicy.RUNTIME)@Target({ElementType.TYPE})public @interface SPI { /** * default extension name / String value() default “”;}@SPIpublic interface Travel { public void travel();}配置文件采用key=value的形式配置car=com.demo.spi.jdk.CarTravelplane=com.demo.spi.jdk.PlaneTravel测试duubo spi加载public class DubboSPITest { public static void main(String[] args) { ExtensionLoader<Travel> extensionLoader = ExtensionLoader.getExtensionLoader(Travel.class); Travel travel = extensionLoader.getExtension(“car”); travel.travel(); }}输出结果travel by carDubbo SPI实现1.扩展类的加载dubbo spi 是通过 ExtensionLoader 类来实现的，通过ExtensionLoader获取接口对应类型的扩展加载器ExtensionLoader.getExtensionLoader,第一次获取会新创建然后会缓存到 ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS 中，之后就在缓存中取就可以了 //获取扩展加载器ExtensionLoader,type必须是@SPI注解接口 //ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS 缓存 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 an interface!”); } if (!withExtensionAnnotation(type)) { throw new IllegalArgumentException(“Extension type (” + type + “) is not an extension, because it is NOT annotated with @” + SPI.class.getSimpleName() + “!”); } 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; }构造方法private ExtensionLoader(Class<?> type) { this.type = type; objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension()); }获取到接口对应的ExtensionLoader后，通过getExtension(String name) 方法获取name对应的扩展对象public T getExtension(String name) { if (StringUtils.isEmpty(name)) { throw new IllegalArgumentException(“Extension name == null”); } if (“true”.equals(name)) { return getDefaultExtension(); } //根据名称获取holder对象，如果不存在则创建一个holder //holder里缓存了name对应的实现类 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) { //1.getExtensionClasses() 获取所有扩展类 //根据name获取扩展类的Class Class<?> clazz = getExtensionClasses().get(name); if (clazz == null) { throw findException(name); } try { T instance = (T) EXTENSION_INSTANCES.get(clazz); if (instance == null) { //2.通过反射newInstance()创建扩展类实例，放到EXTENSION_INSTANCES 缓存中 EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance()); instance = (T) EXTENSION_INSTANCES.get(clazz); } //3.依赖注入 IOC injectExtension(instance); Set<Class<?>> wrapperClasses = cachedWrapperClasses; //4.循环创建 Wrapper 实例 AOP 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); } }获取所有Extension getExtensionClassesprivate Map<String, Class<?>> getExtensionClasses() { Map<String, Class<?>> classes = cachedClasses.get(); if (classes == null) { synchronized (cachedClasses) { classes = cachedClasses.get(); if (classes == null) { //通过loadExtensionClasses 获取所有扩展类，并放到 //cachedClasses 缓存中 classes = loadExtensionClasses(); cachedClasses.set(classes); } } } return classes; }loadExtensionClassesprivate Map<String, Class<?>> loadExtensionClasses() { //缓存默认的扩展名 cacheDefaultExtensionName(); Map<String, Class<?>> extensionClasses = new HashMap<>(); //去dubbo spi所在的路径下META-INF/services/，META-INF/services/internal，META-INF/dubbo/ // 加载扩展类，放到extensionClasses 中 loadDirectory(extensionClasses, DUBBO_INTERNAL_DIRECTORY, type.getName()); loadDirectory(extensionClasses, DUBBO_INTERNAL_DIRECTORY, type.getName().replace(“org.apache”, “com.alibaba”)); loadDirectory(extensionClasses, DUBBO_DIRECTORY, type.getName()); loadDirectory(extensionClasses, DUBBO_DIRECTORY, type.getName().replace(“org.apache”, “com.alibaba”)); loadDirectory(extensionClasses, SERVICES_DIRECTORY, type.getName()); loadDirectory(extensionClasses, SERVICES_DIRECTORY, type.getName().replace(“org.apache”, “com.alibaba”)); return extensionClasses; }loadDirectory->loadResource->loadClass 最终的类加载实现private void loadClass(Map<String, Class<?>> extensionClasses, java.net.URL resourceURL, Class<?> clazz, String name) throws NoSuchMethodException { if (!type.isAssignableFrom(clazz)) { throw new IllegalStateException(“Error occurred when loading extension class (interface: " + type + “, class line: " + clazz.getName() + “), class " + clazz.getName() + " is not subtype of interface.”); } // 如果类有@Adaptive 注解，缓存到cachedAdaptiveClass if (clazz.isAnnotationPresent(Adaptive.class)) { cacheAdaptiveClass(clazz); } else if (isWrapperClass(clazz)) { //如果类的构造方法有加载类的类型，缓存到Set<Class<?>> cachedWrapperClasses; 实现aop的类 cacheWrapperClass(clazz); } else { //如果是普通类 clazz.getConstructor(); //获取扩展类name，为空就取默认name if (StringUtils.isEmpty(name)) { name = findAnnotationName(clazz); if (name.length() == 0) { throw new IllegalStateException(“No such extension name for the class " + clazz.getName() + " in the config " + resourceURL); } } String[] names = NAME_SEPARATOR.split(name); if (ArrayUtils.isNotEmpty(names)) { cacheActivateClass(clazz, names[0]); for (String n : names) { //缓存到cachedNames cacheName(clazz, n); //存储到extensionClasses saveInExtensionClass(extensionClasses, clazz, name); } } } }2. dubbo ioc 注入实现dubbo ioc aop 利用setter注入方式实现private T injectExtension(T instance) { try { if (objectFactory != null) { for (Method method : instance.getClass().getMethods()) { //判断是否是setter方法 if (isSetter(method)) { /* * Check {@link DisableInject} to see if we need auto injection for this property */ if (method.getAnnotation(DisableInject.class) != null) { continue; } Class<?> pt = method.getParameterTypes()[0]; if (ReflectUtils.isPrimitives(pt)) { continue; } try { String property = getSetterProperty(method); //从objectFactory 获取依赖对象 Object object = objectFactory.getExtension(pt, property); if (object != null) { //setter注入 method.invoke(instance, object); } } catch (Exception e) { logger.error(“Failed to inject via method " + method.getName() + " of interface " + type.getName() + “: " + e.getMessage(), e); } } } } } catch (Exception e) { logger.error(e.getMessage(), e); } return instance; }3. dubbo aop 实现//遍历所有Wrapper类创建 Wrapper 实例 //setter注入injectExtension if (CollectionUtils.isNotEmpty(wrapperClasses)) { for (Class<?> wrapperClass : wrapperClasses) { instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance)); } }Dubbo 扩展点自适应机制1. 什么自适应扩展自适应扩展的标记注解@Documented@Retention(RetentionPolicy.RUNTIME)@Target({ElementType.TYPE, ElementType.METHOD})public @interface Adaptive { String[] value() default {};}Adaptive 标记在类上，表示扩展点的加载是人工编码完成，目前只有两个类被@Adaptive， AdaptiveCompiler、AdaptiveExtensionFactoryAdaptive 标记在方法上，表示扩展点的加载是代理完成AdaptiveCompiler 示例@Adaptivepublic class AdaptiveCompiler implements Compiler { private static volatile String DEFAULT_COMPILER; public static void setDefaultCompiler(String compiler) { DEFAULT_COMPILER = compiler; } @Override public Class<?> compile(String code, ClassLoader classLoader) { Compiler compiler; ExtensionLoader<Compiler> loader = ExtensionLoader.getExtensionLoader(Compiler.class); String name = DEFAULT_COMPILER; // copy reference if (name != null && name.length() > 0) { compiler = loader.getExtension(name); } else { compiler = loader.getDefaultExtension(); } return compiler.compile(code, classLoader); }}2. 加载自适应扩展dubbo通过ExtensionLoader.getAdaptiveExtension() 获取自适应扩展实例public T getAdaptiveExtension() { Object instance = cachedAdaptiveInstance.get(); if (instance == null) { if (createAdaptiveInstanceError == null) { synchronized (cachedAdaptiveInstance) { instance = cachedAdaptiveInstance.get(); if (instance == null) { try { //创建自适应扩展，并缓存到cachedAdaptiveInstance instance = createAdaptiveExtension(); cachedAdaptiveInstance.set(instance); } catch (Throwable t) { createAdaptiveInstanceError = t; throw new IllegalStateException(“Failed to create adaptive instance: " + t.toString(), t); } } } } else { throw new IllegalStateException(“Failed to create adaptive instance: " + createAdaptiveInstanceError.toString(), createAdaptiveInstanceError); } } return (T) instance; }createAdaptiveExtension()方法代码private T createAdaptiveExtension() { try { //反射创建AdaptiveExtensionClass，并注入依赖 return injectExtension((T) getAdaptiveExtensionClass().newInstance()); } catch (Exception e) { throw new IllegalStateException(“Can’t create adaptive extension " + type + “, cause: " + e.getMessage(), e); } }获取Class 类 getAdaptiveExtensionClass()private Class<?> getAdaptiveExtensionClass() { getExtensionClasses(); if (cachedAdaptiveClass != null) { return cachedAdaptiveClass; } //创建自适应扩展类 return cachedAdaptiveClass = createAdaptiveExtensionClass(); }createAdaptiveExtensionClass() 生成自适应扩展类private Class<?> createAdaptiveExtensionClass() { //1.生成自适应扩展类代码 String code = new AdaptiveClassCodeGenerator(type, cachedDefaultName).generate(); ClassLoader classLoader = findClassLoader(); //2.获取编译器 org.apache.dubbo.common.compiler.Compiler compiler = ExtensionLoader.getExtensionLoader(org.apache.dubbo.common.compiler.Compiler.class).getAdaptiveExtension(); //3.编译生成class return compiler.compile(code, classLoader); }Protocol 自适应扩展类生成代码 Protocol$Adaptivepackage org.apache.dubbo.rpc;import org.apache.dubbo.common.extension.ExtensionLoader;public class Protocol$Adaptive implements org.apache.dubbo.rpc.Protocol { public void destroy() { throw new UnsupportedOperationException(“The method public abstract void org.apache.dubbo.rpc.Protocol.destroy() of interface org.apache.dubbo.rpc.Protocol is not adaptive method!”); } public int getDefaultPort() { throw new UnsupportedOperationException(“The method public abstract int org.apache.dubbo.rpc.Protocol.getDefaultPort() of interface org.apache.dubbo.rpc.Protocol is not adaptive method!”); } public org.apache.dubbo.rpc.Invoker refer(java.lang.Class arg0, org.apache.dubbo.common.URL arg1) throws org.apache.dubbo.rpc.RpcException { if (arg1 == null) throw new IllegalArgumentException(“url == null”); org.apache.dubbo.common.URL url = arg1; 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.refer(arg0, arg1); } 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); }}