Dubbo源码解析之SPI

柔情只为你懂 2022-05-16 15:47 270阅读 0赞

# 准备 #

dubbo版本：2.5.4

# Dubbo SPI源码解析过程 #

`Dubbo` 在服务发布过程中默认会加载自适应的协议扩展，在类 `ServiceConfig` 中存在以下初始化代码，下面以此进行SPI过程分析。

private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

相关注解

*  `@SPI`：表示当前这个接口是一个扩展点，可以实现自己的扩展实现，例如 `Protocol` 接口默认扩展点实现是 `DubboProtocol`。
 *  `@Adaptive`：表示一个自适应扩展点，在方法级别上，会动态生成一个适配器类。

getExtensionLoader

该方法需要一个 `Class` 类型的参数，该参数表示希望加载的扩展点类型，该参数必须是接口，且该接口必须被 `@SPI` 注解注释，否则拒绝处理。检查通过之后首先会检查 `ExtensionLoader` 缓存中是否已经存在该扩展对应的 `ExtensionLoader` ，如果有则直接返回，否则创建一个新的 `ExtensionLoader` 负责加载该扩展实现，同时将其缓存起来。可以看到对于每一个扩展，`dubbo` 中只会有一个对应的 `ExtensionLoader` 实例

// ExtensionLoader
    	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;
        }

getAdaptiveExtension

// ExtensionLoader
        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;
        }

createAdaptiveExtension

// ExtensionLoader
    private T createAdaptiveExtension() { 
        try { 
            // 1.首先通过getAdaptiveExtensionClass获取适配器扩展点Class并创建实例对象
            // 2.通过injectExtension实现扩展点注入
            return injectExtension((T) getAdaptiveExtensionClass().newInstance());
        } catch (Exception e) { 
            throw new IllegalStateException("Can not create adaptive extenstion " + type + ", cause: " + e.getMessage(), e);
        }
    }

getAdaptiveExtensionClass

// ExtensionLoader
    private Class<?> getAdaptiveExtensionClass() { 
        // 加载所有路径下的扩展点>>
        getExtensionClasses();
        // 判断当前扩展点(Protocol)是否存在自定义的适配器，有则直接返回，没有则动态创建
        if (cachedAdaptiveClass != null) { 
            return cachedAdaptiveClass;
        }
        // 动态生成扩展点适配器类>>
        return cachedAdaptiveClass = createAdaptiveExtensionClass();
    }

getExtensionClasses

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

loadExtensionClasses

private Map<String, Class<?>> loadExtensionClasses() { 
        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<?>>();
        // 加载META-INF/dubbo、META-INF/dubbo/internal、META-INF/services 三个目录下的扩展点实现类
        loadFile(extensionClasses, DUBBO_INTERNAL_DIRECTORY);
        loadFile(extensionClasses, DUBBO_DIRECTORY);
        loadFile(extensionClasses, SERVICES_DIRECTORY);
        return extensionClasses;
    }

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.");
                                            }
                                            /** * 判断是否有自定义适配器类 * Adaptive注解加在类上，说明当前类就是自适应扩展点 * 加在方法上，则需要动态创建一个自适应扩展点(Protocol则对应Protocol$Adaptive) */
                                            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);
        }
    }

回到 `getAdaptiveExtensionClass` 中的 `createAdaptiveExtensionClass`方法

// ExtensionLoader
    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);
    }

回到 `createAdaptiveExtension` 中的 `injectExtension` 扩展点注入方法

private T injectExtension(T instance) { 
            try { 
                if (this.objectFactory != null) { 
                    Method[] var2 = instance.getClass().getMethods();
                    int var3 = var2.length;
    
                    for(int var4 = 0; var4 < var3; ++var4) { 
                        Method method = var2[var4];
                        // 判断是否以set开头，通过setter进行动态注入
                        if (method.getName().startsWith("set") && method.getParameterTypes().length == 1 && Modifier.isPublic(method.getModifiers())) { 
                            // 获取setter方法的参数类型
                            Class pt = method.getParameterTypes()[0];
    
                            try { 
                                // 获取属性名称
                                String property = method.getName().length() > 3 ? method.getName().substring(3, 4).toLowerCase() + method.getName().substring(4) : "";
                                // 根绝类型及名称获取对应的扩展点
                                Object object = this.objectFactory.getExtension(pt, property);
                                if (object != null) { 
                                    method.invoke(instance, object);
                                }
                            } catch (Exception var9) { 
                                logger.error("fail to inject via method " + method.getName() + " of interface " + this.type.getName() + ": " + var9.getMessage(), var9);
                            }
                        }
                    }
                }
            } catch (Exception var10) { 
                logger.error(var10.getMessage(), var10);
            }
    
            return instance;
        }

至此，整个扩展点加载过程完成。