浅析Dubbo的SPI扩展机制
之前可以先看下Dubbo官方文档对SPI的介绍Dubbo SPI。 我们来看下官方文档是怎么说的：
SPI 全称为 Service Provider Interface，是一种服务发现机制。SPI 的本质是将接口实现类的全限定名配置在文件中，并由服务加载器读取配置文件，加载实现类。这样可以在运行时，动态为接口替换实现类。正因此特性，我们可以很容易的通过 SPI 机制为我们的程序提供拓展功能。
SPI 机制在第三方框架中也有所应用，比如 Dubbo 就是通过 SPI 机制加载所有的组件。不过，Dubbo 并未使用 Java 原生的 SPI 机制，而是对其进行了增强，使其能够更好的满足需求。
从官方示例说起
Dubbo SPI 的相关逻辑被封装在了 ExtensionLoader 类中，通过 ExtensionLoader，我们可以加载指定的实现类。Dubbo SPI 所需的配置文件需放置在 META-INF/dubbo 路径下，文件名是扩展接口的全限定名。例如com.alibaba.dubbo.rpc.Filter的文件名就是com.alibaba.dubbo.rpc.Filter。
注意：这里的配置文件是放在你自己的 jar 包内，不是 dubbo 本身的 jar 包内，Dubbo 会全 ClassPath 扫描所有 jar 包内同名的这个文件，然后进行合并。
官方例子的配置内容如下。
optimusPrime = org.apache.spi.OptimusPrime bumblebee = org.apache.spi.Bumblebee 复制代码
以下是示例代码:
public class DubboSPITest {
@Test
public void sayHello() throws Exception {
ExtensionLoader<Robot> extensionLoader =
ExtensionLoader.getExtensionLoader(Robot.class);
Robot optimusPrime = extensionLoader.getExtension("optimusPrime");
optimusPrime.sayHello();
Robot bumblebee = extensionLoader.getExtension("bumblebee");
bumblebee.sayHello();
}
}
复制代码
实现分析
获取ExtensionLoad
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<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;
}
复制代码
这里的EXTENSION_LOADERS是一个静态变量。
/** * 扩展加载器集合 */ private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<Class<?>, ExtensionLoader<?>>(); 复制代码
注意到withExtensionAnnotation方法，每个要扩展的接口都必须要包含@SPI。
private static <T> boolean withExtensionAnnotation(Class<T> type) {
return type.isAnnotationPresent(SPI.class);
}
复制代码
我们看下ExtensionLoader的构造函数。
private ExtensionLoader(Class<?> type) {
this.type = type;
objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension());
}
复制代码
此时这里的type是上面例子传入的Robot.class。
获取拓展对象
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;
}
复制代码
可以看到如果如果name是true,则返回默认的扩展对象。 检查缓存是否命中，没有的话就调用createExtension方法创建扩展对象。
查找默认的扩展对象
public T getDefaultExtension() {
getExtensionClasses();
if (null == cachedDefaultName || cachedDefaultName.length() == 0
|| "true".equals(cachedDefaultName)) {
return null;
}
return getExtension(cachedDefaultName);
}
复制代码
先调用getExtensionClasses方法获取所有的扩展实现类，再获取cachedDefaultName对应的扩展对象。那cachedDefaultName的值又是哪里来的呢？其实就是SPI注解的value。在loadExtensionClasses方法中有这段代码。
private Map<String, Class<?>> loadExtensionClasses() {
// 通过@SPI注解,获得默认的扩展实现类名
final SPI defaultAnnotation = type.getAnnotation(SPI.class);
if (defaultAnnotation != null) {
String value = defaultAnnotation.value();
if ((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];
}
}
}
// 省略方法代码
return extensionClasses;
}
复制代码
创建扩展对象
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);
// 创建Wrapper扩展对象
Set<Class<?>> wrapperClasses = cachedWrapperClasses;
if (wrapperClasses != null && !wrapperClasses.isEmpty()) {
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 + ") could not be instantiated: " + t.getMessage(), t);
}
}
复制代码
createExtension方法包括以下的几个步骤:
- 通过 getExtensionClasses 获取所有的拓展类
- 通过反射创建拓展对象
- 向拓展对象中注入依赖
- 将拓展对象包裹在相应的 Wrapper 对象中
获取所有的扩展类
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进行双重检查，若缓存的扩展实现类集合为空，则调用loadExtensionClasses方法加载扩展实现类集合。
private Map<String, Class<?>> loadExtensionClasses() {
// 通过@SPI注解,获得默认的扩展实现类名
final SPI defaultAnnotation = type.getAnnotation(SPI.class);
if (defaultAnnotation != null) {
String value = defaultAnnotation.value();
if ((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<?>>();
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;
}
复制代码
加载的配置文件路径见下面。
private static final String SERVICES_DIRECTORY = "META-INF/services/"; private static final String DUBBO_DIRECTORY = "META-INF/dubbo/"; private static final String DUBBO_INTERNAL_DIRECTORY = DUBBO_DIRECTORY + "internal/"; 复制代码
从配置文件中，加载扩展实现类数组。
private void loadDirectory(Map<String, Class<?>> extensionClasses, String dir, String type) {
String fileName = dir + type;
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 resourceURL = urls.nextElement();
loadResource(extensionClasses, classLoader, resourceURL);
}
}
} catch (Throwable t) {
logger.error("Exception when load extension class(interface: " +
type + ", description file: " + fileName + ").", t);
}
}
复制代码
private void loadResource(Map<String, Class<?>> extensionClasses, ClassLoader classLoader, java.net.URL resourceURL) {
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(resourceURL.openStream(), "utf-8"));
try {
String line;
while ((line = reader.readLine()) != null) {
// 跳过当前被注释调的情况，例如 #spring=xxxxxxxx
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) {
loadClass(extensionClasses, resourceURL, Class.forName(line, true, classLoader), name);
}
} 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);
}
}
}
} finally {
reader.close();
}
} catch (Throwable t) {
logger.error("Exception when load extension class(interface: " +
type + ", class file: " + resourceURL + ") in " + resourceURL, t);
}
}
复制代码
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 when load extension class(interface: " +
type + ", class line: " + clazz.getName() + "), class "
+ clazz.getName() + "is not subtype of interface.");
}
// 缓存自适应扩展对象的类别 'cachedAdaptiveClass'
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());
}
// 缓存拓展Wrapper实现类到'cachedWrapperClasses'
} else if (isWrapperClass(clazz)) {
Set<Class<?>> wrappers = cachedWrapperClasses;
if (wrappers == null) {
cachedWrapperClasses = new ConcurrentHashSet<Class<?>>();
wrappers = cachedWrapperClasses;
}
wrappers.add(clazz);
} else {
// 缓存扩展实现类'extensionClasses'
clazz.getConstructor();
if (name == null || name.length() == 0) {
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 (names != null && names.length > 0) {
Activate activate = clazz.getAnnotation(Activate.class);
if (activate != null) {
cachedActivates.put(names[0], activate);
} else {
// support com.alibaba.dubbo.common.extension.Activate
com.alibaba.dubbo.common.extension.Activate oldActivate = clazz.getAnnotation(com.alibaba.dubbo.common.extension.Activate.class);
if (oldActivate != null) {
cachedActivates.put(names[0], oldActivate);
}
}
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());
}
}
}
}
}
复制代码
向扩展对象中注入依赖
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())) {
/**
* 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];
try {
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);
}
} 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;
}
复制代码
遍历查找set方法，反射注入依赖。
扩展点自适应
参考官方文档的自适应扩展机制。
有关Adaptive注解还可参考 Dubbo SPI之Adaptive详解 。
在类上加Adaptive注解优先级最高。
private Class<?> getAdaptiveExtensionClass() {
getExtensionClasses();
if (cachedAdaptiveClass != null) {
return cachedAdaptiveClass;
}
return cachedAdaptiveClass = createAdaptiveExtensionClass();
}
复制代码
在loadClass方法中可以看到带有Adaptive注解的类对象会被赋值给cachedAdaptiveClass。
// 缓存自适应扩展对象的类别 'cachedAdaptiveClass'
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());
}
// 缓存拓展Wrapper实现类到'cachedWrapperClasses'
}
复制代码
SPI注解的value值是是默认值，在上面我们知道被赋值给了cachedDefaultName，如果通过url获取不到取哪个类作为Adaptive类的话就用这个默认值。
String defaultExtName = cachedDefaultName;
String getNameCode = null;
for (int i = value.length - 1; i >= 0; --i) {
if (i == value.length - 1) {
if (null != defaultExtName) {
if (!"protocol".equals(value[i])) {
if (hasInvocation) {
getNameCode = String.format("url.getMethodParameter(methodName, /"%s/", /"%s/")", value[i], defaultExtName);
} else {
getNameCode = String.format("url.getParameter(/"%s/", /"%s/")", value[i], defaultExtName);
}
} else {
getNameCode = String.format("( url.getProtocol() == null ? /"%s/" : url.getProtocol() )", defaultExtName);
}
} else {
if (!"protocol".equals(value[i])) {
if (hasInvocation) {
getNameCode = String.format("url.getMethodParameter(methodName, /"%s/", /"%s/")", value[i], defaultExtName);
} else {
getNameCode = String.format("url.getParameter(/"%s/")", value[i]);
}
} else {
getNameCode = "url.getProtocol()";
}
}
} else {
if (!"protocol".equals(value[i])) {
if (hasInvocation) {
getNameCode = String.format("url.getMethodParameter(methodName, /"%s/", /"%s/")", value[i], defaultExtName);
} else {
getNameCode = String.format("url.getParameter(/"%s/", %s)", value[i], getNameCode);
}
} else {
getNameCode = String.format("url.getProtocol() == null ? (%s) : url.getProtocol()", getNameCode);
}
}
}
复制代码
扩展点自动激活
对于一些集合类扩展点，比如：Filter，可以同时加载多个实现，此时可以用自动激活来简化配置。
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Filter;
@Activate // 无条件自动激活
public class XxxFilter implements Filter {
// ...
}
复制代码
或者
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Filter;
@Activate("xxx") // 当配置了xxx参数，并且参数为有效值时激活，比如配了cache="lru"，自动激活CacheFilter。
public class XxxFilter implements Filter {
// ...
}
复制代码
或者
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Filter;
@Activate(group = "provider", value = "xxx") // 只对提供方激活，group可选"provider"或"consumer"
public class XxxFilter implements Filter {
// ...
}
复制代码
或者
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Filter;
@Activate(group = "provider", value = "xxx", order=1) // order表示优先级，低的排序优先级高
public class XxxFilter implements Filter {
// ...
}
复制代码
我们来看下实现
public List<T> getActivateExtension(URL url, String key, String group) {
String value = url.getParameter(key);
return getActivateExtension(url, value == null || value.length() == 0 ? null : Constants.COMMA_SPLIT_PATTERN.split(value), group);
}
复制代码
public List<T> getActivateExtension(URL url, String[] values, String group) {
List<T> exts = new ArrayList<T>();
List<String> names = values == null ? new ArrayList<String>(0) : Arrays.asList(values);
// 处理自动激活的扩展对象们
// 判断不存在配置 "-name"
if (!names.contains(Constants.REMOVE_VALUE_PREFIX + Constants.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)) {
T ext = getExtension(name);
if (!names.contains(name)
&& !names.contains(Constants.REMOVE_VALUE_PREFIX + name)
&& isActive(activateValue, url)) {
exts.add(ext);
}
}
}
Collections.sort(exts, ActivateComparator.COMPARATOR);
}
// 处理自定义配置的扩展对象们
List<T> usrs = new ArrayList<T>();
for (int i = 0; i < names.size(); i++) {
String name = names.get(i);
if (!name.startsWith(Constants.REMOVE_VALUE_PREFIX)
&& !names.contains(Constants.REMOVE_VALUE_PREFIX + name)) {
// 将配置的自定义在自动激活的扩展对象们前面
if (Constants.DEFAULT_KEY.equals(name)) {
if (!usrs.isEmpty()) {
exts.addAll(0, usrs);
usrs.clear();
}
} else {
T ext = getExtension(name);
usrs.add(ext);
}
}
}
if (!usrs.isEmpty()) {
exts.addAll(usrs);
}
return exts;
}
复制代码
正文到此结束
- 本文标签： constant HashMap tk apache find 需求 key 自适应 Service 集合类 HashSet java tar 代码 Collections consumer Property 本质 stream spring IDE URLs provider message id 缓存 https UI 遍历 dubbo ACE Collection synchronized struct 配置 App final http IO equals ip map CTO 参数 src cache ArrayList classpath cat CEO list lib ORM ConcurrentHashMap ssl description 注释 value
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
