Apache Dubbo 是一款微服务开发框架,它提供了 RPC通信 与 微服务治理 两大关键能力。这意味着,使用 Dubbo 开发的微服务,将具备相互之间的远程发现与通信能力, 同时利用 Dubbo 提供的丰富服务治理能力,可以实现诸如服务发现、负载均衡、流量调度等服务治理诉求。同时 Dubbo 是高度可扩展的,用户几乎可以在任意功能点去定制自己的实现,以改变框架的默认行为来满足自己的业务需求
本文主要给大家讲解下Dubbo的扩展点原理。
JDK中的SPI(Service Provider Interface)提供了一种基于接口的扩展机制,主要实现步骤如下:
具体不清楚的可以参考下我的另一篇专门介绍SPI的文章: Java SPI内容详解 学习交流 463257262
在Dubbo中的扩展点主要是对JDK的扩展点思想做了增强,主要增强了一下功能:
helloService=com.bobo.spring.cloud.alibaba.consumer.spi.impl.HelloServiceImpl
HelloService HelloService = ExtensionLoader
.getExtensionLoader(HelloService.class)
.getExtension("helloService");
System.out.println(HelloService.sayHello("wangxing"));
在Dubbo中存在了以下三种类型的扩展点(Q群:463257262):
在Dubbo中,通过在接口上标注【@SPI】标识该接口是一个扩展点,同时在其扩展点实现类或方法上,如果存在【@Adaptive】注解,则表示该类或方法是一个自适应的扩展点。标注在类上时,表示该扩展类是默认的自适应扩展点,标注在方法上时,表示该方法是自适应扩展点,将会重写该方法,使得Dubbo能够在运行时获取到具体的扩展点。加下来就进入源码的分析吧…
Dubbo的扩展点的入口如下:
HelloService helloService2 = ExtensionLoader
.getExtensionLoader(HelloService.class)
.getAdaptiveExtension();
首先进入到getExtensionLoader()方法
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;
}
该方法主要做了以下几件事
接下来进入ExtensionLoader类中的构造方法,看看ExtensionLoader实例化时做了什么
private ExtensionLoader(Class<?> type) {
this.type = type;
objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension());
}
在构造器中为type属性复制传入的扩展点的.class对象。同时通过自适应扩展点的方式获取到了一个ExtensionFactory的扩展点的实现,赋值给objectFactory。这里先不详细说明会具体获取到哪个实现,本节分析完成再回过来看。
通过以上的步骤,一个初始化完成的ExtensionLoader对象已经被获取到了,分析getAdaptiveExtension()获取自适应扩展点实现的流程
进入getAdaptiveExtension()
public T getAdaptiveExtension() {
Object instance = cachedAdaptiveInstance.get();
if (instance == null) {
// 如果获取到的实例为null,切缓存的错误不能null,抛出异常
if (createAdaptiveInstanceError != null) {
throw new IllegalStateException("Failed to create adaptive instance: " +
createAdaptiveInstanceError.toString(),
createAdaptiveInstanceError);
}
synchronized (cachedAdaptiveInstance) {
instance = cachedAdaptiveInstance.get();
if (instance == null) {
try {
instance = createAdaptiveExtension();
cachedAdaptiveInstance.set(instance);
} catch (Throwable t) {
// 异常信息缓存起来,下一次进来时如果发现是创建实例是出现异常,就直接抛出异常。这里的设计应该是当扩展点创建异常时避免多次执行创建流程的优化
createAdaptiveInstanceError = t;
throw new IllegalStateException("Failed to create adaptive instance: " + t.toString(), t);
}
}
}
}
return (T) instance;
}
该方法主要做了以下几件事
创建扩展点实现的具体流程是在createAdaptiveExtension方法中
private Class<?> getAdaptiveExtensionClass() {
getExtensionClasses();
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;
}
该方法首先会从cachedClasses(cachedClasses也是一个holder,用于存储每个扩展点的所有扩展实现的map集合)获取该.class对象,存在则直接返回,否则调用loadExtensionClasses方法加载扩展点的classs,将加载到的classes存到cachedClasses中。 接下来进入loadExtensionClasses
private Map<String, Class<?>> loadExtensionClasses() {
cacheDefaultExtensionName();
Map<String, Class<?>> extensionClasses = new HashMap<>();
for (LoadingStrategy strategy : strategies) {
// 扫描每个加载策略目录中的扩展点实现
loadDirectory(extensionClasses, strategy.directory(), type.getName(), strategy.preferExtensionClassLoader(), strategy.overridden(), strategy.excludedPackages());
// 对alibaba的践行兼容
loadDirectory(extensionClasses, strategy.directory(), type.getName().replace("org.apache", "com.alibaba"), strategy.preferExtensionClassLoader(), strategy.overridden(), strategy.excludedPackages());
}
return extensionClasses;
}
首先调用cacheDefaultExtensionName()方法。再给接口标注@SPI的时候,可以给个默认的value值,表示指定的默认的扩展点实现,如@SPI(“dubbo”)public interface Protocol 表示默认的扩展点为扩展点实现名为dubbo的实现。而cacheDefaultExtensionName方法就是通过注解获取到该扩展点的默认扩展点name,赋值给cachedDefaultName
遍历strategies,获取到多个LoadingStrategy,通过stratery.directory()获取到需要扫描的目录,以下是Dubbo中默认的三种策略的实现
在Dubbo中,创建ExtensionLoader对象时,会load到所有的LoadingStrategy,这里利用的是JDK原生的SPI的方式,将LoadingStrategy的所有扩展实现都加载进来,保存到strategies中。所以如果需要扩展Dubbo中的扫描的路径,按照JDK的原生方式进行扩展即可
进入到loadDirectory方法
private void loadDirectory(Map<String, Class<?>> extensionClasses, String dir, String type,
boolean extensionLoaderClassLoaderFirst, boolean overridden, String... excludedPackages) {
String fileName = dir + type;
try {
Enumeration<java.net.URL> urls = null;
ClassLoader classLoader = findClassLoader();
// try to load from ExtensionLoader's ClassLoader first
if (extensionLoaderClassLoaderFirst) {
ClassLoader extensionLoaderClassLoader = ExtensionLoader.class.getClassLoader();
if (ClassLoader.getSystemClassLoader() != extensionLoaderClassLoader) {
urls = extensionLoaderClassLoader.getResources(fileName);
}
}
if (urls == null || !urls.hasMoreElements()) {
if (classLoader != null) {
urls = classLoader.getResources(fileName);
} else {
urls = ClassLoader.getSystemResources(fileName);
}
}
if (urls != null) {
while (urls.hasMoreElements()) {
java.net.URL resourceURL = urls.nextElement();
loadResource(extensionClasses, classLoader, resourceURL, overridden, excludedPackages);
}
}
} catch (Throwable t) {
logger.error("Exception occurred when loading extension class (interface: " +
type + ", description file: " + fileName + ").", t);
}
}
该方法首先扫描传入路径下的所有的以type全类名命名的文件,获取到资源,将获取到的文件转换成Resource,传入到loadResource()方法中
private void loadResource(Map<String, Class<?>> extensionClasses, ClassLoader classLoader,
java.net.URL resourceURL, boolean overridden, String... excludedPackages) {
try {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(resourceURL.openStream(), StandardCharsets.UTF_8))) {
String line;
// 按行读取文件中的每行数据
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;
// 通过等号分割,等号前的为key(扩展点name),等号后的为类的全类名
int i = line.indexOf('=');
if (i > 0) {
name = line.substring(0, i).trim();
line = line.substring(i + 1).trim();
}
if (line.length() > 0 && !isExcluded(line, excludedPackages)) {
loadClass(extensionClasses, resourceURL, Class.forName(line, true, classLoader), name, overridden);
}
} catch (Throwable t) {
IllegalStateException e = new IllegalStateException("Failed to load extension class (interface: " + type + ", class line: " + line + ") in " + resourceURL + ", cause: " + t.getMessage(), t);
exceptions.put(line, e);
}
}
}
}
} catch (Throwable t) {
logger.error("Exception occurred when loading extension class (interface: " +
type + ", class file: " + resourceURL + ") in " + resourceURL, t);
}
}
首先按行读取文件,对读取到的每一行数据通过=号进行分割,=号前为扩展点的名字,=号后的为扩展点的具体扩展的实现类的全类名
通过Class.forName将实现类加载到内存中。传入到loadClass()方法
private void loadClass(Map<String, Class<?>> extensionClasses, java.net.URL resourceURL, Class<?> clazz, String name,
boolean overridden) 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.");
}
if (clazz.isAnnotationPresent(Adaptive.class)) {
// 会覆盖掉之前保存的`cachedAdaptiveClass`
cacheAdaptiveClass(clazz, overridden);
} else if (isWrapperClass(clazz)) {
cacheWrapperClass(clazz);
} else {
clazz.getConstructor();
if (StringUtils.isEmpty(name)) {
// 如果扩展文件中的name为空,则调用findAnnotationName方法获取扩展点名字,具体命名方式这里就不详细看了
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) {
cacheName(clazz, n);
saveInExtensionClass(extensionClasses, clazz, n, overridden);
}
}
}
}
该方法主要主要做了以下几件事
到这里,该扩展点在项目中的所有实现将被加载完成,且已经区分出了实现中,自适应扩展点,wrapper等不同类型的实现。然后我们回到
再次回到getAdaptiveExtensionClass()方法,当执行完getExtensionClasses();方法之后,如果cacheAdaptiveClass
为null,表示该扩展点没有默认的自适应扩展点,此时扩展点需要将需要自适应扩展的方法上标注@Adaptive(),并且该方法中需要传入URL对象,因为Dubbo中需要将都是通过URL来携带配置的。
将调用createAdaptiveExtensionClass()方法动态创建一个自适应扩展点
private Class<?> createAdaptiveExtensionClass() {
String code = new AdaptiveClassCodeGenerator(type, cachedDefaultName).generate();
ClassLoader classLoader = findClassLoader();
org.apache.dubbo.common.compiler.Compiler compiler = ExtensionLoader.getExtensionLoader(org.apache.dubbo.common.compiler.Compiler.class).getAdaptiveExtension();
return compiler.compile(code, classLoader);
}
该方法会动态生成一个自适应扩展点的类,然后编译通过编译器编译,加载其.class文件到内存中。返回该动态类的.class对象。
生成的动态类代码如下:
import org.apache.dubbo.common.extension.ExtensionLoader;
public class HelloService$Adaptive implements com.wangx.spring.cloud.alibaba.consumer.spi.HelloService {
public java.lang.String sayHello(org.apache.dubbo.common.URL arg0, java.lang.String arg1) {
if (arg0 == null) throw new IllegalArgumentException("url == null");
org.apache.dubbo.common.URL url = arg0;
// 默认类名分割。可以在@Adaptive注解中指定该参数的名称,default为SPI上定义的默认扩展点实现
String extName = url.getParameter("hello.service","default");
if (extName == null)
throw new IllegalStateException("Failed to get extension (com.wangx.spring.cloud.alibaba.consumer.spi.HelloService) name from url (" + url.toString() + ") use keys([hello.service])");
// 根据名称获取到该扩展点类型的扩展实现
com.wangx.spring.cloud.alibaba.consumer.spi.HelloService extension = (com.wangx.spring.cloud.alibaba.consumer.spi.HelloService) ExtensionLoader.getExtensionLoader(com.wangx.spring.cloud.alibaba.consumer.spi.HelloService.class).getExtension(extName);
return extension.sayHello(arg0, arg1);
}
}
该类实现了 扩展点接口,重写了扩展点中的自适应扩展方法。该方法体现的是,在运行时通过传入的URL的信息动态的获取处理当前URL时的扩展点的实现。
到这里就可以返回各种情况下的自适应扩展点的.class对象了,接下来再次回到createAdaptiveExtension方法中,通过以上的一系列操作,我们已经获取到了自适应扩展点的.class对象,并调用反射创建一个扩展点实现的对象。然后调用injectExtension进行依赖注入
private T injectExtension(T instance) {
if (objectFactory == null) {
return instance;
}
try {
for (Method method : instance.getClass().getMethods()) {
if (!isSetter(method)) {
continue;
}
/**
* 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);
Object object = objectFactory.getExtension(pt, property);
if (object != null) {
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;
}
在injectExtension方法中,有如下几个操作:
在初始化ExtensionLoader对象时,objectFactory是通过如下代码获取的
objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension());
通过上面的一些列分析,现在已经可以知道这是一个ExtensionFactory的自适应的扩展点,根据我们自适应扩展点的获取方式,我们可以推断出该扩展点的自适应扩展点的实现。
在ExtensionFactory的所有子类实现中,我们找到了AdaptiveExtensionFactory类,该类上标注了@Adaptive,所以可以推断出objectFacotory的指向的就是AdaptiveExtensionFactory类的对象。
下面来看看AdaptiveExtensionFactory类中的实现:
@Adaptive
public 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);
}
@Override
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;
}
}
在AdaptiveExtensionFactory的构造函数中,会先获取到ExtensionFactory类型的ExtensionLoader,然后通过调用loader.getSupportedExtensions()获取到ExtensionFactory的所有扩展实现的名字。通过loader.getExtension(name)根据名称获取到所有扩展点,存储到factories中。
在injectExtension中调用getExtension()方法时,将会遍历初始化时获取到的所有的ExtensionFactory的扩展点。只要在其中的一个扩展到那点中找到该扩展对象的实例,则直接返回。传入的对象type和name,获取Dubbo环境下可能存在的扩展点。
在injectExtension方法中返回的依赖注入完成的对象,即是我们需要获取的自适应扩展点对象
根据名称获取扩展点,顾名思义,就是根据扩展点的名称,获取到扩展点对应的实现。这种方式在Dubbo中也被广泛应用到,主要是可以通过URL中的参数或协议作为name,在运行时根据URl动态的获取到不同方式的实现。比如获取负载均衡器等
入口如下:
HelloService HelloService = ExtensionLoader.getExtensionLoader(HelloService.class).getExtension("helloService");
getExtensionLoader方法在上述中已经解释清楚了,现在直接进入到getExtension方法中
public T getExtension(String name) {
//判断传入名称是否为null
if (StringUtils.isEmpty(name)) {
throw new IllegalArgumentException("Extension name == null");
}
// 如果为true,获取默认的扩展点,在getDefaultExtension方法中会调用getExtensionClasses->loadExtensionClasses方法,该方法中的cacheDefaultExtensionName会将默认扩展点的name赋值到cachedDefaultName中,所以当调用getDefaultExtension()即可获得默认的扩展点实现
if ("true".equals(name)) {
return getDefaultExtension();
}
final 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;
}
该方法主要做了以下几个操作
创建instance时,需要调用createExtension(name)方法
@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, clazz.newInstance());
instance = (T) EXTENSION_INSTANCES.get(clazz);
}
injectExtension(instance);
Set<Class<?>> wrapperClasses = cachedWrapperClasses;
if (CollectionUtils.isNotEmpty(wrapperClasses)) {
for (Class<?> wrapperClass : wrapperClasses) {
instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));
}
}
initExtension(instance);
return instance;
} catch (Throwable t) {
throw new IllegalStateException("Extension instance (name: " + name + ", class: " +
type + ") couldn't be instantiated: " + t.getMessage(), t);
}
}
通过自适应扩展点的分析,我们已经知道了getExtensionClasses()方法返回的是扩展点name为key,value为具体扩展点实现类的class对象。然后可以通过name获取到扩展点对应的class对象。然后通过该class对象,到EXTENSION_INSTANCES缓存中获取实现类的对象,如果不存在,则直接通过反射创建一个对象。
接下来就跟自适应扩展点一样,调用injectExtension进行依赖注入。依赖注入完成之后,将对该对象进行包装,首先从加载时装载的cachedWrapperClasses缓存中获取所有的wrapper扩展点,遍历所有的装饰器,将创建的实际的扩展点对象通过构造器传入到wrapper中,反射创建出一个wrapper对象,在对该wrapper对象进行依赖注入,完成之后将该对象复制给instance。
比如现在有一个扩展点S,扩展点实现F,该扩展点有A,B,C三个包装器,那么通过如上遍历包装器之后,最后的到的instance对象的结构可能经过了层层的嵌套,变成了这个样子:A(B(C(F)))。调用instance时,将会从最外层开始执行该对象的方法,最终到最里层才会执行实际扩展点的方法。这种设计使得包装的实现更加简洁和灵活。
然后调用initExtension方法,如果该对象实现了Lifecycle接口,则调用initialize方法。最后返回一个被包装的对象
激活扩展点的使用也需要在实现类上标注@Activate注解。注解中可以指定group和value,当不指定时,就无条件激活,否则就按照指定的条件进行激活,激活扩展点的入口为:
List<HelloServiceActive> li = ExtensionLoader.getExtensionLoader(HelloServiceActive.class).getActivateExtension(url,"helloService2");
for (HelloServiceActive helloServiceActive : li) {
helloServiceActive.say();
}
需要传入一个URL对象和一个需要获取激活的的扩展点的参数看key,比如在参数中设置url = url.addParameter(“xing”,“xing”);传入的key为xing。进入getActivateExtension()方法,因为可以指定group等,所以最终调用的方法为
public List<T> getActivateExtension(URL url, String[] values, String group) {
List<T> activateExtensions = new ArrayList<>();
List<String> names = values == null ? new ArrayList<>(0) : asList(values);
if (!names.contains(REMOVE_VALUE_PREFIX + DEFAULT_KEY)) {
getExtensionClasses();
for (Map.Entry<String, Object> entry : cachedActivates.entrySet()) {
String name = entry.getKey();
Object activate = entry.getValue();
String[] activateGroup, activateValue;
if (activate instanceof Activate) {
activateGroup = ((Activate) activate).group();
activateValue = ((Activate) activate).value();
} else if (activate instanceof com.alibaba.dubbo.common.extension.Activate) {
activateGroup = ((com.alibaba.dubbo.common.extension.Activate) activate).group();
activateValue = ((com.alibaba.dubbo.common.extension.Activate) activate).value();
} else {
continue;
}
if (isMatchGroup(group, activateGroup)
&& !names.contains(name)
&& !names.contains(REMOVE_VALUE_PREFIX + name)
&& isActive(activateValue, url)) {
activateExtensions.add(getExtension(name));
}
}
activateExtensions.sort(ActivateComparator.COMPARATOR);
}
List<T> loadedExtensions = new ArrayList<>();
for (int i = 0; i < names.size(); i++) {
String name = names.get(i);
if (!name.startsWith(REMOVE_VALUE_PREFIX)
&& !names.contains(REMOVE_VALUE_PREFIX + name)) {
if (DEFAULT_KEY.equals(name)) {
if (!loadedExtensions.isEmpty()) {
activateExtensions.addAll(0, loadedExtensions);
loadedExtensions.clear();
}
} else {
loadedExtensions.add(getExtension(name));
}
}
}
if (!loadedExtensions.isEmpty()) {
activateExtensions.addAll(loadedExtensions);
}
return activateExtensions;
}
经过我们对上面两种方式的分析,其实第三种方式我们已经能够很轻松的看懂了。主要流程有,先遍历扫描是装载在cachedActivates中的所有激活扩展点,跟url和group进行匹配,匹配成功,则通过扩展点的名name通过根据名称获取扩展点的方式获取扩展点,存储到list中。然后遍历根据key获取到的value解析出来的扩展点名称,通过该名称获取到扩展点装载到list中,然后返回activateExtensions();
这样就完成了通过激活扩展点的获取。