Dubbo SPI.png
1. 首先是主程序类com.alibaba.dubbo.common.Main，成员变量如下：
public static final String CONTAINER_KEY = "dubbo.container";
public static final String SHUTDOWN_HOOK_KEY = "dubbo.shutdown.hook";
private static final Logger logger = LoggerFactory.getLogger(Main.class);
private static final ExtensionLoader<Container> loader = ExtensionLoader.getExtensionLoader(Container.class); //静态代码中初始化容器的ExtensionLoader
private static final ReentrantLock LOCK = new ReentrantLock();
private static final Condition STOP;
static {
STOP = LOCK.newCondition();
}
public static void main(String[] args) {
try {
if (args == null || args.length == 0) {
String config = ConfigUtils.getProperty("dubbo.container", loader.getDefaultExtensionName());
args = Constants.COMMA_SPLIT_PATTERN.split(config);
}
final List<Container> containers = new ArrayList();
for(int i = 0; i < args.length; ++i) {
containers.add(loader.getExtension(args[i]));
}
logger.info("Use container type(" + Arrays.toString(args) + ") to run dubbo serivce.");
if ("true".equals(System.getProperty("dubbo.shutdown.hook"))) {
Runtime.getRuntime().addShutdownHook(new Thread("dubbo-container-shutdown-hook") {
public void run() {
Iterator i$ = containers.iterator();
while(i$.hasNext()) {
Container container = (Container)i$.next();
try {
container.stop();
Main.logger.info("Dubbo " + container.getClass().getSimpleName() + " stopped!");
} catch (Throwable var8) {
Main.logger.error(var8.getMessage(), var8);
}
try {
Main.LOCK.lock();
Main.STOP.signal();
} finally {
Main.LOCK.unlock();
}
}
}
});
}
Iterator i$ = containers.iterator();
while(i$.hasNext()) {
Container container = (Container)i$.next();
container.start();
logger.info("Dubbo " + container.getClass().getSimpleName() + " started!");
}
System.out.println((new SimpleDateFormat("[yyyy-MM-dd HH:mm:ss]")).format(new Date()) + " Dubbo service server started!");
} catch (RuntimeException var10) {
var10.printStackTrace();
logger.error(var10.getMessage(), var10);
System.exit(1);
}
try {
LOCK.lock();
STOP.await();
} catch (InterruptedException var8) {
logger.warn("Dubbo service server stopped, interrupted by other thread!", var8);
} finally {
LOCK.unlock();
}
}
可以看到Dubbo启动时主要做了以下的事情：
- 从main方法参数或者配置文件的dubbo.container属性中取出容器名称数组
- 使用静态代码中已经初始化好的ExtensionLoader<Container>加载所有容器
- 如果配置中启用了dubbo.shutdown.hook，Dubbo会使用JRE的addShutdownHook钩子绑定新线程，在这个线程中配合信号量同步，实现了kill命令优雅停机
- 逐个启动容器
- 阻塞等待停机信号量
2. 接下来我们看扩展加载器com.alibaba.dubbo.common.extension.ExtensionLoader类：
private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class);
//JavaSPI路径
private static final String SERVICES_DIRECTORY = "META-INF/services/";
//Dubbo外部SPI路径
private static final String DUBBO_DIRECTORY = "META-INF/dubbo/"; private static final String DUBBO_INTERNAL_DIRECTORY = "META-INF/dubbo/internal/";
//Dubbo内部SPI路径
private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*");
//扩展类的ExtensionLoader集合
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap();
//扩展类的实例集合
private static final ConcurrentMap<Class<?>, Object> EXTENSION_INSTANCES = new ConcurrentHashMap();
//要被扩展的接口类
private final Class<?> type;
//用于IOC注入其他扩展类实例的扩展对象工厂
private final ExtensionFactory objectFactory;
//loadExtension或者loadClass的时候缓存的扩展类名，用于快速获取扩展类对应的字符串名
private final ConcurrentMap<Class<?>, String> cachedNames = new ConcurrentHashMap();
//loadExtensionClasses时初始化的Holder，里面是一个HashMap，保存扩展类Class的容器
private final Holder<Map<String, Class<?>>> cachedClasses = new Holder();
//用于缓存各个扩展类的Activate注解
private final Map<String, Activate> cachedActivates = new ConcurrentHashMap();
//用于缓存已经加载的扩展类的实例
private final ConcurrentMap<String, Holder<Object>> cachedInstances = new ConcurrentHashMap();
//用于缓存适配类的实例（每个ExtensionLoader只有一个）
private final Holder<Object> cachedAdaptiveInstance = new Holder();
//缓存适配类，主要用于在并发操作下保证适配类的唯一性
private volatile Class<?> cachedAdaptiveClass = null;
//缓存的默认扩展类的名称
private String cachedDefaultName;
private volatile Throwable createAdaptiveInstanceError;
//一个缓存了所有包装类的ConcurrentHashSet
private Set<Class<?>> cachedWrapperClasses;
private Map<String, IllegalStateException> exceptions = new ConcurrentHashMap();
getExtensionLoader类方法从EXTENSION_LOADERS中获取指定类的EXTENSION_LOADERS，不存在则调用构造函数创建。
public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) {
if (type == null) {
throw new IllegalArgumentException("Extension type == null");
} else if (!type.isInterface()) {
throw new IllegalArgumentException("Extension type(" + type + ") is not interface!");
} else if (!withExtensionAnnotation(type)) {
throw new IllegalArgumentException("Extension type(" + type + ") is not extension, because WITHOUT @" + SPI.class.getSimpleName() + " Annotation!");
} else {
ExtensionLoader<T> loader = (ExtensionLoader)EXTENSION_LOADERS.get(type);
if (loader == null) {
//虽然一般不会出现并发创建同一个类的ExtensionLoader的情况，但是出现时这里会重复创建多个实例，造成资源浪费
EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader(type));
loader = (ExtensionLoader)EXTENSION_LOADERS.get(type);
}
return loader;
}
}
构造函数里会保存当前的扩展类类型，同时对ExtensionFactory自身之外的其他类调用getAdaptiveExtension方法获取对应的适配类。
private ExtensionLoader(Class<?> type) {
this.type = type;
this.objectFactory = type == ExtensionFactory.class ? null : (ExtensionFactory)getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension();
}
以下就是获取扩展点适配类的调用链，而加载扩展点实例的过程则从getExtension开始。
public T getAdaptiveExtension() {
Object instance = this.cachedAdaptiveInstance.get();
if (instance == null) {
if (this.createAdaptiveInstanceError != null) {
throw new IllegalStateException("fail to create adaptive instance: " + this.createAdaptiveInstanceError.toString(), this.createAdaptiveInstanceError);
}
Holder var2 = this.cachedAdaptiveInstance;
synchronized(this.cachedAdaptiveInstance) {
instance = this.cachedAdaptiveInstance.get();
if (instance == null) {
try {
instance = this.createAdaptiveExtension();
this.cachedAdaptiveInstance.set(instance);
} catch (Throwable var5) {
this.createAdaptiveInstanceError = var5;
throw new IllegalStateException("fail to create adaptive instance: " + var5.toString(), var5);
}
}
}
}
return instance;
}
private T createAdaptiveExtension() {
try {
return this.injectExtension(this.getAdaptiveExtensionClass().newInstance());
} catch (Exception var2) {
throw new IllegalStateException("Can not create adaptive extension " + this.type + ", cause: " + var2.getMessage(), var2);
}
}
/*
@Adaptive注解在类上：代表人工实现，实现一个装饰类（设计模式中的装饰模式），它主要作用于固定已知类，
目前整个系统只有2个，AdaptiveCompiler、AdaptiveExtensionFactory。
为什么AdaptiveCompiler这个类是固定已知的？因为整个框架仅支持Javassist和JdkCompiler。
为什么AdaptiveExtensionFactory这个类是固定已知的？因为整个框架仅支持2个objFactory,一个是spi,另一个是spring
*/
private Class<?> getAdaptiveExtensionClass() {
//先加载扩展类，如果扩展类里已经有@Adaptive的实现类，那么会加载并存入cachedAdaptiveClass，否则Dubbo会调用createAdaptiveExtensionClass用字节码引擎动态生成
this.getExtensionClasses();
return this.cachedAdaptiveClass != null ? this.cachedAdaptiveClass : (this.cachedAdaptiveClass = this.createAdaptiveExtensionClass());
}
private Class<?> createAdaptiveExtensionClass() {
String code = this.createAdaptiveExtensionClassCode();
ClassLoader classLoader = findClassLoader();
Compiler compiler = (Compiler)getExtensionLoader(Compiler.class).getAdaptiveExtension();
return compiler.compile(code, classLoader);
}
private String createAdaptiveExtensionClassCode() {
//这里用StringBuilder拼出了扩展类的代码，比较冗长，故此处省略
}
private T injectExtension(T instance) {
try {
if (this.objectFactory != null) {
Method[] arr$ = instance.getClass().getMethods();
int len$ = arr$.length;
for(int i$ = 0; i$ < len$; ++i$) {
Method method = arr$[i$];
if (method.getName().startsWith("set") && method.getParameterTypes().length == 1 && Modifier.isPublic(method.getModifiers()) && method.getAnnotation(DisableInject.class) == null) {
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;
}
public T getExtension(String name) {
if (name != null && name.length() != 0) {
if ("true".equals(name)) {
return this.getDefaultExtension();
} else {
Holder<Object> holder = (Holder)this.cachedInstances.get(name);
if (holder == null) {
this.cachedInstances.putIfAbsent(name, new Holder());
holder = (Holder)this.cachedInstances.get(name);
}
Object instance = holder.get();
if (instance == null) {
synchronized(holder) {
instance = holder.get();
if (instance == null) {
instance = this.createExtension(name);
holder.set(instance);
}
}
}
return instance;
}
} else {
throw new IllegalArgumentException("Extension name == null");
}
private T createExtension(String name) {
Class<?> clazz = (Class)this.getExtensionClasses().get(name);
if (clazz == null) {
throw this.findException(name);
} else {
try {
T instance = EXTENSION_INSTANCES.get(clazz);
if (instance == null) {
EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance());
instance = EXTENSION_INSTANCES.get(clazz);
}
this.injectExtension(instance);
Set<Class<?>> wrapperClasses = this.cachedWrapperClasses;
Class wrapperClass;
if (wrapperClasses != null && !wrapperClasses.isEmpty()) {
for(Iterator i$ = wrapperClasses.iterator(); i$.hasNext(); instance = this.injectExtension(wrapperClass.getConstructor(this.type).newInstance(instance))) {
wrapperClass = (Class)i$.next();
}
}
return instance;
} catch (Throwable var7) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + this.type + ") could not be instantiated: " + var7.getMessage(), var7);
}
}
}
private Map<String, Class<?>> getExtensionClasses() {
Map<String, Class<?>> classes = (Map)this.cachedClasses.get();
if (classes == null) {
Holder var2 = this.cachedClasses;
synchronized(this.cachedClasses) {
classes = (Map)this.cachedClasses.get();
if (classes == null) {
classes = this.loadExtensionClasses();
this.cachedClasses.set(classes);
}
}
}
return classes;
}
private Map<String, Class<?>> loadExtensionClasses() {
SPI defaultAnnotation = (SPI)this.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 " + this.type.getName() + ": " + Arrays.toString(names));
}
if (names.length == 1) {
this.cachedDefaultName = names[0];
}
}
}
Map<String, Class<?>> extensionClasses = new HashMap();
this.loadDirectory(extensionClasses, "META-INF/dubbo/internal/");
this.loadDirectory(extensionClasses, "META-INF/dubbo/");
this.loadDirectory(extensionClasses, "META-INF/services/");
return extensionClasses;
}
private void loadDirectory(Map<String, Class<?>> extensionClasses, String dir) {
String fileName = dir + this.type.getName();
try {
ClassLoader classLoader = findClassLoader();
Enumeration urls;
if (classLoader != null) {
urls = classLoader.getResources(fileName);
} else {
urls = ClassLoader.getSystemResources(fileName);
}
if (urls != null) {
while(urls.hasMoreElements()) {
java.net.URL resourceURL = (java.net.URL)urls.nextElement();
this.loadResource(extensionClasses, classLoader, resourceURL);
}
}
} catch (Throwable var7) {
logger.error("Exception when load extension class(interface: " + this.type + ", description file: " + fileName + ").", var7);
}
}
private void loadResource(Map<String, Class<?>> extensionClasses, ClassLoader classLoader, java.net.URL resourceURL) {
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(resourceURL.openStream(), "utf-8"));
String line;
try {
while((line = reader.readLine()) != null) {
int ci = line.indexOf(35);
if (ci >= 0) {
line = line.substring(0, ci);
}
line = line.trim();
if (line.length() > 0) {
try {
String name = null;
int i = line.indexOf(61);
if (i > 0) {
name = line.substring(0, i).trim();
line = line.substring(i + 1).trim();
}
if (line.length() > 0) {
this.loadClass(extensionClasses, resourceURL, Class.forName(line, true, classLoader), name);
}
} catch (Throwable var13) {
IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: " + this.type + ", class line: " + line + ") in " + resourceURL + ", cause: " + var13.getMessage(), var13);
this.exceptions.put(line, e);
}
}
}
} finally {
reader.close();
}
} catch (Throwable var15) {
logger.error("Exception when load extension class(interface: " + this.type + ", class file: " + resourceURL + ") in " + resourceURL, var15);
}
}
private void loadClass(Map<String, Class<?>> extensionClasses, java.net.URL resourceURL, Class<?> clazz, String name) throws NoSuchMethodException {
if (!this.type.isAssignableFrom(clazz)) {
throw new IllegalStateException("Error when load extension class(interface: " + this.type + ", class line: " + clazz.getName() + "), class " + clazz.getName() + "is not subtype of interface.");
} else {
if (clazz.isAnnotationPresent(Adaptive.class)) {
if (this.cachedAdaptiveClass == null) {
this.cachedAdaptiveClass = clazz;
} else if (!this.cachedAdaptiveClass.equals(clazz)) {
throw new IllegalStateException("More than 1 adaptive class found: " + this.cachedAdaptiveClass.getClass().getName() + ", " + clazz.getClass().getName());
}
} else if (this.isWrapperClass(clazz)) {
Set<Class<?>> wrappers = this.cachedWrapperClasses;
if (wrappers == null) {
this.cachedWrapperClasses = new ConcurrentHashSet();
wrappers = this.cachedWrapperClasses;
}
wrappers.add(clazz);
} else {
clazz.getConstructor();
if (name == null || name.length() == 0) {
name = this.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 = (Activate)clazz.getAnnotation(Activate.class);
if (activate != null) {
this.cachedActivates.put(names[0], activate);
}
String[] arr$ = names;
int len$ = names.length;
for(int i$ = 0; i$ < len$; ++i$) {
String n = arr$[i$];
if (!this.cachedNames.containsKey(clazz)) {
this.cachedNames.put(clazz, n);
}
Class<?> c = (Class)extensionClasses.get(n);
if (c == null) {
extensionClasses.put(n, clazz);
} else if (c != clazz) {
throw new IllegalStateException("Duplicate extension " + this.type.getName() + " name " + n + " on " + c.getName() + " and " + clazz.getName());
}
}
}
}
}
}
总结这段代码，可以发现：
- 几乎所有的Dubbo类都是从SPI机制加载来的
- 最先被加载的扩展类是ExtensionFactory，然后是Compiler，接着是其适配类，最后才是Container
- IOC注入的实现在方法injectExtension中
- AOP机制是通过包装类来实现的，而Dubbo判断是否为包装类的标准是是否有带被包装类参数的构造函数
private boolean isWrapperClass(Class<?> clazz) {
try {
clazz.getConstructor(this.type);
return true;
} catch (NoSuchMethodException var3) {
return false;
}
}
- 动态类的本质是可以做到一个SPI中的不同的Adaptive方法可以去调不同的SPI实现类去处理。使得程序的灵活性大大提高。这才是整套SPI设计的一个精华之所在
此外对于集合类扩展点，比如：Filter, InvokerListener, ExportListener, TelnetHandler, StatusChecker等， @Adaptive可以同时加载多个实现，此时，可以用自动激活来简化配置
public List<T> getActivateExtension(URL url, String[] values, String group) {
List<T> exts = new ArrayList();
List<String> names = values == null ? new ArrayList(0) : Arrays.asList(values);
String name;
if (!((List)names).contains("-default")) {
this.getExtensionClasses();
Iterator i$ = this.cachedActivates.entrySet().iterator();
while(i$.hasNext()) {
Entry<String, Activate> entry = (Entry)i$.next();
name = (String)entry.getKey();
Activate activate = (Activate)entry.getValue();
if (this.isMatchGroup(group, activate.group())) {
T ext = this.getExtension(name);
if (!((List)names).contains(name) && !((List)names).contains("-" + name) && this.isActive(activate, url)) {
exts.add(ext);
}
}
}
Collections.sort(exts, ActivateComparator.COMPARATOR);
}
List<T> usrs = new ArrayList();
for(int i = 0; i < ((List)names).size(); ++i) {
name = (String)((List)names).get(i);
if (!name.startsWith("-") && !((List)names).contains("-" + name)) {
if ("default".equals(name)) {
if (!usrs.isEmpty()) {
exts.addAll(0, usrs);
usrs.clear();
}
} else {
T ext = this.getExtension(name);
usrs.add(ext);
}
}
}
if (!usrs.isEmpty()) {
exts.addAll(usrs);
}
return exts;
}
实际和AdaptiveExtension一样在解析时，如果有Activate注解就会放到cachedActivates里。
以dubbo中的filter为例子，ProtocolFilterWrapper中的代码，调用getActivateExtension方法获得激活的filters，然后依次执行，所谓的集合类的扩展点其实dubbo自身代码已经决定了的，调用形式不同而已
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
Invoker<T> last = invoker;
// 获取配置的filters
List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
if (filters.size() > 0) {
for (int i = filters.size() - 1; i >= 0; i --) {
final Filter filter = filters.get(i);
final Invoker<T> next = last;
last = new Invoker<T>() {
public Class<T> getInterface() {
return invoker.getInterface();
}
public URL getUrl() {
return invoker.getUrl();
}
public boolean isAvailable() {
return invoker.isAvailable();
}
public Result invoke(Invocation invocation) throws RpcException {
return filter.invoke(next, invocation);
}
public void destroy() {
invoker.destroy();
}
@Override
public String toString() {
return invoker.toString();
}
};
}
}
return last;
}
3. 其他
对比Java SPI：
*不会在应用程序启动时加载所有扩展点实例
*可以使用注解指定要加载的实例
*如果加载失败会抛出可识别可追踪的异常
*支持IOC和AOP机制，同时兼容Spring Bean
不同方式配置SPI实现类的优先级：
实现类Adaptive注解 > URL参数 > SPI注解
网友评论