Dubbo SPI机制详解
程序员文章站
2022-06-01 19:46:33
...
本篇其实更应该在Dubbo系列的最早去写，只不过因为当初在读源码的时候没有开始更新博客，所以漏了重要的一个章节，那就是Dubbo的SPI机制，所以这一章节主要是讲解一下Dubbo的SPI机制实现。
我们都是知道一个合格的开源框架对于扩展的支持都要是相当弹性的，Dubbo 也不例外。Dubbo采用了简单的扩展方式，基于SPI机制。
什么是SPI？
java spi的具体约定为:当服务的提供者，提供了服务接口的一种实现之后，在jar包的META-INF/services/目录里同时创建一个以服务接口命名的文件。该文件里就是实现该服务接口的具体实现类。而当外部程序装配这个模块的时候，就能通过该jar包META-INF/services/里的配置文件找到具体的实现类名，并装载实例化，完成模块的注入。 基于这样一个约定就能很好的找到服务接口的实现类，而不需要再代码里制定。jdk提供服务实现查找的一个工具类：java.util.ServiceLoader。 （网上摘录的解释）
Dubbo的SPI主要改进了JDK标准的SPI实现：
JDK标准的SPI会一次性实例化扩展点所有实现，如果有扩展实现初始化很耗时，但如果没用上也加载，会很浪费资源。
如果扩展点加载失败，连扩展点的名称都拿不到了。比如：JDK标准的ScriptEngine，通过getName();获取脚本类型的名称，但如果RubyScriptEngine因为所依赖的jruby.jar不存在，导致RubyScriptEngine类加载失败，这个失败原因被吃掉了，和ruby对应不起来，当用户执行ruby脚本时，会报不支持ruby，而不是真正失败的原因。
增加了对扩展点IoC和AOP的支持，一个扩展点可以直接setter注入其它扩展点。
基本上讲到这里大家对于SPI可能有个大致的认识，但是要真正理解Dubbo的SPI，还是要仔细看一下源码才可以。
在理解Dubbo的SPI之前，要明确几个核心概念：
扩展点 Dubbo作用灵活的框架，并不会强制所有用户都一定使用Dubbo提供的某些架构。例如注册中心（Registry），Dubbo提供了zk和redis，但是如果我们更倾向于其他的注册中心的话，我们可以替换掉Dubbo提供的注册中心。针对这种可被替换的技术实现点我们称之为扩展点，类似的扩展点有很多，例如Protocol，Filter，Loadbalance等等。
Wrapper Dubbo在加载某个接口的扩展类时候，如果某个实现中有一个拷贝类构造函数，那么该接口实现就是该接口的包装类，此时Dubbo会在真正的实现类上层包装上盖Wrapper。即这个时候从ExtensionLoader中返回的实际扩展类是被Wrapper包装的接口实现类。
Adaptive 这个自适应的扩展点比较难理解，所以这里直接以一个例子来讲解：在RegistryProtocol中有一个属性为Cluster，其中Protocol和Cluster都是Dubbo提供的扩展点，所以这时候当我们真正在操作中使用cluster的时候究竟使用的哪一个cluster的实现类呢？是FailbackCluster还是FailoverCluster？Dubbo在加载一个扩展点的时候如果发现其成员变量也是一个扩展点并且有相关的set方法，就会在这时候将该扩展点设置为一个自适应的扩展点，自适应扩展点（Adaptive）会在真正使用的时候从URL中获取相关参数，来调用真正的扩展点实现类。具体的实现会在下面的源码中详细解释。对于Adaptive的理解其实个人推荐的是Dubbo开发者指南，指南中有对于Adaptive的明确介绍。
Activate 官网的叫法是自**，其实这个更合适的叫法我认为是条件**，我们还记得上一篇中有提到Filter的内容，其中Filter链的获取就是通过@Activate注解来确定的，所以Activate的作用主要是：提供一种选择性**的条件，可以是我们通过相关的配置来确定**哪些功能。
ExtensionLoader
之所以把ExtensionLoader标这么大是因为其太重要，作为整个SPI的核心，ExtensionLoader起着无可替代的作用，下面的整篇文章都在围绕着这个类进行讲解，足以看出他是多么重要了。
鉴于ExtensionLoade的用法比较多的都是如下用法，我就以下面的调用为例开始介绍ExtensionLoader （调用的链路比较长，大家要耐心点哈）
//在Dubbo源码中大面积使用这种写法，都是获得某个接口的适配类，在真正执行的时候才决定最终的作用类
private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
//因为每一个扩展类加载器只能加载特定的SPI扩展的实现，所以要获得某个扩展的实现的话首先要找到他对应的扩展类加载器(ExtensionLoader)
//一个扩展接口的所有实现都是被同一个扩展类加载器来加载的
@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!");
}
//获得某个扩展类加载器的时候该接口必须被@SPI修饰才可以
if(!withExtensionAnnotation(type)) {
throw new IllegalArgumentException("Extension type(" + type +
") is not extension, because WITHOUT @" + SPI.class.getSimpleName() + " Annotation!");
}
// 先从静态缓存中获取对应的ExtensionLoader实例，每个接口对应一个ExtensionLoader并且把映射关系都存储在缓存之中
ExtensionLoader<T> loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
if (loader == null) {
// 为Extension类型创建ExtensionLoader实例，并放入静态缓存
EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader<T>(type));
loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
}
return loader;
}
public T getAdaptiveExtension() {
Object instance = cachedAdaptiveInstance.get();
if (instance == null) {
if(createAdaptiveInstanceError == null) {
//采用单例模式的双重判断，这种模式感觉使用的范围都比较广泛
//cachedAdaptiveInstance作为一个Holder（只有简单的get和set方法），也是一个锁对象
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;
}
// 创建一个接口的适配类
private T createAdaptiveExtension() {
try {
//获取AdaptiveExtensionClass并完成注入
//基本分两步：1.获取适配器类 2.在适配器里面注入其他的扩展点
return injectExtension((T) getAdaptiveExtensionClass().newInstance());
} catch (Exception e) {
throw new IllegalStateException("Can not create adaptive extenstion " + type + ", cause: " + e.getMessage(), e);
}
}
//获得适配类有两种途径，第一就是某个实现类上被@Adaptive注解，第二就是没有实现类被注解，因此Dubbo会自动生成一个某个接口的适配类
private Class<?> getAdaptiveExtensionClass() {
//如果能找到被@Adaptive注解实现类
getExtensionClasses();
if (cachedAdaptiveClass != null) {
return cachedAdaptiveClass;
}
//找不到的话就自行创建一个适配类
return cachedAdaptiveClass = createAdaptiveExtensionClass();
}
//加载当前扩展所有实现，看是否有实现类上被标注为@Adaptive
private Map<String, Class<?>> getExtensionClasses() {
//多次判断是为了防止同一个扩展点被多次加载
Map<String, Class<?>> classes = cachedClasses.get();
if (classes == null) {
synchronized (cachedClasses) {
classes = cachedClasses.get();
if (classes == null) {
//loadExtensionClasses会加载所有的配置文件，将配置文件中对应的的类加载到当前的缓存中
//load完之后该classes已经保留了所有的扩展类映射关系
classes = loadExtensionClasses();
cachedClasses.set(classes);
}
}
}
return classes;
}
private Map<String, Class<?>> loadExtensionClasses() {
//所有的扩展点接口都必须被SPI注释标注
final SPI defaultAnnotation = type.getAnnotation(SPI.class);
if(defaultAnnotation != null) {
String value = defaultAnnotation.value();
if(value != null && (value = value.trim()).length() > 0) {
//一个@SPI注解的值只能有一个
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));
}
//cachedDefaultName表示该扩展点对应的默认适配类的key
//逻辑运行到这里就意味着该扩展点有定义的适配类，不需要Dubbo框架自己生成适配类
if(names.length == 1) cachedDefaultName = names[0];
}
}
Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>();
//加载对应目录下的配置文件，三个目录分别为：META-INF/services/，META-INF/dubbo，META-INF/dubbo/internal
loadFile(extensionClasses, DUBBO_INTERNAL_DIRECTORY);
loadFile(extensionClasses, DUBBO_DIRECTORY);
loadFile(extensionClasses, SERVICES_DIRECTORY);
return extensionClasses;
}
//加载相关路径下的类文件
private void loadFile(Map<String, Class<?>> extensionClasses, String dir) {
String fileName = dir + type.getName();
try {
Enumeration<java.net.URL> urls;
//加载这些问价你的classloader要和加载当前类的classloader一致,这个类似与Java默认的类加载器和类的加载关系
ClassLoader classLoader = findClassLoader();
if (classLoader != null) {
//该步骤就加载所有的classpath下面的同名文件（包含你的项目本地classpath和依赖jar包）
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();//SPI扩展文件中的key
line = line.substring(i + 1).trim();//SPI扩展文件中配置的value ExtensionLoader是根据key和value同时加载的
}
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注解的话
if (clazz.isAnnotationPresent(Adaptive.class)) {
if(cachedAdaptiveClass == null) {
//将缓存的AdaptiveClass设置成此类
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);
}
}
}
//将配置的key名字根据逗号来区分
String[] names = NAME_SEPARATOR.split(name);
if (names != null && names.length > 0) {
Activate activate = clazz.getAnnotation(Activate.class);
if (activate != null) {
//自**的实现类只会将第一个name进行存储，原因到现在还有点不清楚
cachedActivates.put(names[0], activate);
}
for (String n : names) {
//假如说配置了name1,name2=com.alibaba.DemoInterface,这时候在cachedNames中只会存储一个name1
if (! cachedNames.containsKey(clazz)) {
cachedNames.put(clazz, n);
}
Class<?> c = extensionClasses.get(n);
if (c == null) {
extensionClasses.put(n, clazz);
//防止同一个扩展类实现被两个key共同使用
} 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);
}
}
//通过反射自动调用instance的set方法把自身的属性注入进去，解决的扩展类依赖问题，也就是说解决扩展类依赖扩展类的问题
private T injectExtension(T instance) {
try {
if (objectFactory != null) {
for (Method method : instance.getClass().getMethods()) {
if (method.getName().startsWith("set")
&& method.getParameterTypes().length == 1
&& Modifier.isPublic(method.getModifiers())) { //如果该扩展点实例有Set开头的公共方法
Class<?> pt = method.getParameterTypes()[0];//得到set方法的参数类型
try {
//得到属性名称，比如setName方法就得到name属性名称
String property = method.getName().length() > 3 ? method.getName().substring(3, 4).toLowerCase() + method.getName().substring(4) : "";
Object object = objectFactory.getExtension(pt, property);//从获得属性值
if (object != null) {
method.invoke(instance, object);// 如果不为空，说明set方法的参数是扩展点类型，那么进行注入，意思也就是说扩展点里面还有依赖其他扩展点
}
} catch (Exception e) {
logger.error("fail to inject via method " + method.getName()
+ " of interface " + type.getName() + ": " + e.getMessage(), e);
}
}
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return instance;
}
//通过反射自动调用instance的set方法把自身的属性注入进去，解决的扩展类依赖问题，也就是说解决扩展类依赖扩展类的问题
private T injectExtension(T instance) {
try {
if (objectFactory != null) {
for (Method method : instance.getClass().getMethods()) {
if (method.getName().startsWith("set")
&& method.getParameterTypes().length == 1
&& Modifier.isPublic(method.getModifiers())) { //如果该扩展点实例有Set开头的公共方法
Class<?> pt = method.getParameterTypes()[0];//得到set方法的参数类型
try {
//得到属性名称，比如setName方法就得到name属性名称
String property = method.getName().length() > 3 ? method.getName().substring(3, 4).toLowerCase() + method.getName().substring(4) : "";
Object object = objectFactory.getExtension(pt, property);//从获得属性值
if (object != null) {
method.invoke(instance, object);// 如果不为空，说明set方法的参数是扩展点类型，那么进行注入，意思也就是说扩展点里面还有依赖其他扩展点
}
} catch (Exception e) {
logger.error("fail to inject via method " + method.getName()
+ " of interface " + type.getName() + ": " + e.getMessage(), e);
}
}
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return instance;
}
基本上讲到之类，一个适配类已经可以被加载出来了，但是由于上面的逻辑比较深，下面给出简单文字逻辑：
1.为了获得一个扩展点的适配类，首先会看缓存中有没有已经加载过的适配类，如果有的话就直接返回，没有的话就进入第2步。
2.加载所有的配置文件，将所有的配置类都load进内存并且在ExtensionLoader内部做好缓存，如果配置的文件中有适配类就缓存起来，如果没有适配类就自行通过代码自行创建适配类并且缓存起来（代码之后给出样例）。
3.在加载配置文件的时候，会依次将包装类，自**的类都进行缓存。
4.将获取完适配类时候，如果适配类的set方法对应的属性也是扩展点话，会依次注入对应的属性的适配类（循环进行）。
讲到这里的话适配类这段代码逻辑已经大致都解释过了，下面看一下Dubbo自己生成的适配类代码是怎样的(以Protocol为例)：
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class Protocol$Adpative implements Protocol {
public Invoker refer(Class arg0, URL arg1) throws Class {
if (arg1 == null) throw new IllegalArgumentException("url == null");
URL url = arg1;
String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
if(extName == null) throw new IllegalStateException("Fail to get extension(Protocol) name from url(" + url.toString() + ") use keys([protocol])");
Protocol extension = (Protocol)ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(extName);
return extension.refer(arg0, arg1);
}
public Exporter export(Invoker arg0) throws Invoker {
if (arg0 == null) throw new IllegalArgumentException("Invoker argument == null");
if (arg0.getUrl() == null) throw new IllegalArgumentException("Invoker argument getUrl() == null");URL url = arg0.getUrl();
//这里会根据url中的信息获取具体的实现类名
String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
if(extName == null) throw new IllegalStateException("Fail to get extension(Protocol) name from url(" + url.toString() + ") use keys([protocol])");
//根据上面的实现类名，会在运行时，通过Dubbo的扩展机制加载具体实现类
Protocol extension = (Protocol)ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(extName);
return extension.export(arg0);
}
public void destroy() {
throw new UnsupportedOperationException("method public abstract void Protocol.destroy() of interface Protocol is not adaptive method!");
}
public int getDefaultPort() {
throw new UnsupportedOperationException("method public abstract int Protocol.getDefaultPort() of interface Protocol is not adaptive method!");
}
}
本质上的做法就是通过方法的参数获得URL信息，从URL中获得对应的value对应值，然后从ExtensionLoader的缓存中找到value对应的具体实现类，然后用该实现类进行工作。可以看到上面的核心就是getExtension方法了，下面来看一下该方法的实现：
public T getExtension(String name) {
if (name == null || name.length() == 0)
throw new IllegalArgumentException("Extension name == null");
//DefaultExtension就是自适应的扩展类
if ("true".equals(name)) {
return getDefaultExtension();
}
//先从缓存中去取
Holder<Object> holder = cachedInstances.get(name);
if (holder == null) {
//如果缓存中没有的话在创建一个然后放进去，但是此时并没有实际内容，只有一个空的容器Holder
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;
}
private T createExtension(String name) {
//获取name类型对应的扩展类型，从cachedClasses根据key获取对应的class，cachedClasses已经在load操作的时候初始化过了。
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方法的作用就是通过set方法注入其他的属性扩展点，上面已经讲过
injectExtension(instance);
Set<Class<?>> wrapperClasses = cachedWrapperClasses;
if (wrapperClasses != null && wrapperClasses.size() > 0) {
for (Class<?> wrapperClass : wrapperClasses) {
//循环遍历所有wrapper实现，实例化wrapper并进行扩展点注入
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);
}
}
getExtension的步骤就是根据之前load操作生成的class缓存，找到对应的Class信息，然后根据Class生成具体的实例，生成之后通过反射注入其他的属性扩展点，将包装类进行层层包装，最终返回包装过的类。（包装逻辑可以参考Protocol的包装实例）
理解ExtensionLoader是理解Dubbo的SPI的基础，也是理解Dubbo扩展机制的基础，所以如果各位有耐心的话，还是请自己阅读一些上面提到的源码部分（基本上都有文字注释）。
最后，习惯性的撒花～～～
上一篇： dubbo系列之dubbo SPI
下一篇： dubbo之SPI