2、SOFARPC源码解析——SPI篇

SOFA RPC 源码解析

1、SOFA RPC 源码解析 —— 服务发布篇

从SOFA RPC 源码解析 —— 服务发布篇中来看有很多地方我都提到了SPI，那么什么是SPI呢，我们简单介绍下JAVA的SPI流程：JAVA的SPI运行流程是运用java.util.ServiceLoader这个类的load方法去在src/META-INF/services/寻找对应的全路径接口名称的文件，然后在文件中找到对应的实现方法并注入实现，然后你可以使用了。（点击JAVA SPI），话不多说，我们有了大概的spi的概念后就来看下SOFA RPC是怎么实现SPI的。

我们就从熟悉的代码开始吧：

public class RpcServer {
public static void main(String[] args) {
// 构建RegistryConfig 注册配置
RegistryConfig registryCOnfig= new RegistryConfig().setProtocol("zookeeper").setAddress("127.0.0.1:2181");
RegistryConfig registryConfig1 = new RegistryConfig().setProtocol("zookeeper").setAddress("127.0.0.1:2181");
List registryCOnfigs= new ArrayList();
registryConfigs.add(registryConfig);
registryConfigs.add(registryConfig1);
// 构建ServerConfig 服务配置
List serverCOnfigs= new ArrayList();
ServerConfig serverCOnfig= new ServerConfig().setProtocol("bolt").setPort(12200).setDaemon(false);
ServerConfig serverConfig1 = new ServerConfig().setProtocol("rest").setPort(12200).setDaemon(false);
serverConfigs.add(serverConfig);
serverConfigs.add(serverConfig1);
// 构建发布配置
ProviderConfig providerCOnfig= new ProviderConfig().setApplication(new ApplicationConfig().setAppName("paul")).setInterfaceId(HelloService.class.getName()).setRef(new HelloServiceImpl()).setServer(serverConfigs).setRegistry(registryConfig);
// 正式发布
providerConfig.export();
}
}

上面那个代码就是我第一篇SOFA RPC 源码解析 —— 服务发布篇里面的，

我们再次简单分解下，就从第一步构建RegistryConfig 注册配置里面用到的SPI来讲：

点进去RegistryConfig的父类AbstractIdConfig

/**
* 默认配置带ID
*
* @param the sub class of AbstractIdConfig
* @author GengZhang
*/
public abstract class AbstractIdConfig implements Serializable {
private static final long serialVersiOnUID= -1932911135229369183L;
/**
* Id生成器
*/
private final static AtomicInteger ID_GENERATOR = new AtomicInteger(0);
static {
RpcRuntimeContext.now();
}
...

再点进去静态模块里的RpcRuntimeContext，找到：

static {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Welcome! Loading SOFA RPC Framework : {}, PID is:{}", Version.BUILD_VERSION, PID);
}
put(RpcConstants.CONFIG_KEY_RPC_VERSION, Version.RPC_VERSION);
// 初始化一些上下文
initContext();
// 初始化其它模块
ModuleFactory.installModules();
// 增加jvm关闭事件
if (RpcConfigs.getOrDefaultValue(RpcOptions.JVM_SHUTDOWN_HOOK, true)) {
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("SOFA RPC Framework catch JVM shutdown event, Run shutdown hook now.");
}
destroy(false);
}
}, "SOFA-RPC-ShutdownHook"));
}
}

这里面的 ModuleFactory.installModules();就用到了SPI，根据配置加载扩展模块，一起来看看吧：

/**
* 加载全部模块
*/
public static void installModules() {
ExtensionLoader loader = ExtensionLoaderFactory.getExtensionLoader(Module.class);
String moduleLoadList = RpcConfigs.getStringValue(RpcOptions.MODULE_LOAD_LIST);
for (Map.Entry> o : loader.getAllExtensions().entrySet()) {
String moduleName = o.getKey();
Module module = o.getValue().getExtInstance();
// judge need load from rpc option
if (needLoad(moduleLoadList, moduleName)) {
// judge need load from implement
if (module.needLoad()) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Install Module: {}", moduleName);
}
module.install();
INSTALLED_MODULES.put(moduleName, module);
} else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("The module " + moduleName + " does not need to be loaded.");
}
}
} else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("The module " + moduleName + " is not in the module load list.");
}
}
}
}

上述代码里,ExtensionLoader loader = ExtensionLoaderFactory.getExtensionLoader(Module.class);里面就是SPI的核心

/**
* Get extension loader by extensible class with listener
*
* @param clazz Extensible class
* @param listener Listener of ExtensionLoader
* @param Class
* @return ExtensionLoader of this class
*/
public static ExtensionLoader getExtensionLoader(Class clazz, ExtensionLoaderListener listener) {
// 第一次进来loader 肯定是空的
ExtensionLoader loader = LOADER_MAP.get(clazz);
if (loader == null) {
// 锁住class,双重校验，防止重复初始化
synchronized (ExtensionLoaderFactory.class) {
loader = LOADER_MAP.get(clazz);
if (loader == null) {
// 实例化 loader
loader = new ExtensionLoader(clazz, listener);
LOADER_MAP.put(clazz, loader);
}
}
}
return loader;
}
/**
* Get extension loader by extensible class without listener
*
* @param clazz Extensible class
* @param Class
* @return ExtensionLoader of this class
*/
public static ExtensionLoader getExtensionLoader(Class clazz) {
return getExtensionLoader(clazz, null);
}

上面我写了简单的注释，最重要的一句话就是loader = new ExtensionLoader(clazz, listener);我们点进去看看它到底做了些什么吧！

/**
* 构造函数（自动加载）
*
* @param interfaceClass 接口类
* @param listener 加载后的监听器
*/
public ExtensionLoader(Class interfaceClass, ExtensionLoaderListener listener) {
this(interfaceClass, true, listener);
}

/**
* 构造函数（主要测试用）
*
* @param interfaceClass 接口类
* @param autoLoad 是否自动开始加载
* @param listener 扩展加载监听器
*/
protected ExtensionLoader(Class interfaceClass, boolean autoLoad, ExtensionLoaderListener listener) {
// 如果RPC框架正在关闭则属性全部初始化为空return
if (RpcRunningState.isShuttingDown()) {
this.interfaceClass = null;
this.interfaceName = null;
this.listener = null;
this.factory = null;
this.extensible = null;
this.all = null;
return;
}
// 接口为空，既不是接口，也不是抽象类，要求必须是接口或者抽象类，会自动加载所有实现类
if (interfaceClass == null ||
!(interfaceClass.isInterface() || Modifier.isAbstract(interfaceClass.getModifiers()))) {
throw new IllegalArgumentException("Extensible class must be interface or abstract class!");
}
this.interfaceClass = interfaceClass;
this.interfaceName = ClassTypeUtils.getTypeStr(interfaceClass);
this.listener = listener;
// 获取extensible注解，上面会有几个属性file指定扩展文件名称，singleton是否单例，coded是否需要编码
Extensible extensible = interfaceClass.getAnnotation(Extensible.class);
if (extensible == null) {
throw new IllegalArgumentException(
"Error when load extensible interface " + interfaceName + ", must add annotation @Extensible.");
} else {
this.extensible = extensible;
}
// 如果是单例的，则存入factory,也就是一个线程安全的ConcurrentHashMap
this.factory = extensible.singleton() ? new ConcurrentHashMap() : null;
// 初始化一个保存全部扩展的对象的ConcurrentMap
this.all = new ConcurrentHashMap>();
// 是否自动加载，一般都是是
if (autoLoad) {
// 从配置中心或者配置文件中加载扩展类文件相对路径
List paths = RpcConfigs.getListValue(RpcOptions.EXTENSION_LOAD_PATH);
for (String path : paths) {
// 这个就是最重要的一步，解析文件！
loadFromFile(path);
}
}
}

/**
* @param path path必须以/结尾
*/
protected synchronized void loadFromFile(String path) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Loading extension of extensible {} from path: {}", interfaceName, path);
}
// 默认如果不指定文件名字，就是接口名
String file = StringUtils.isBlank(extensible.file()) ? interfaceName : extensible.file().trim();
// 获得完整的相对地址
String fullFileName = path + file;
try {
// 获得当前类的类加载器，这个是用来获取resource的也就是获取资源文件
ClassLoader classLoader = ClassLoaderUtils.getClassLoader(getClass());
loadFromClassLoader(classLoader, fullFileName);
} catch (Throwable t) {
if (LOGGER.isErrorEnabled()) {
LOGGER.error("Failed to load extension of extensible " + interfaceName + " from path:" + fullFileName,
t);
}
}
}

protected void loadFromClassLoader(ClassLoader classLoader, String fullFileName) throws Throwable {
// 根据全路径名从classLoader里面获取资源文件
Enumeration urls = classLoader != null ? classLoader.getResources(fullFileName)
: ClassLoader.getSystemResources(fullFileName);
// 可能存在多个文件。
if (urls != null) {
while (urls.hasMoreElements()) {
// 读取一个文件
URL url = urls.nextElement();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Loading extension of extensible {} from classloader: {} and file: {}",
interfaceName, classLoader, url);
}
BufferedReader reader = null;
try {
reader = new BufferedReader(new InputStreamReader(url.openStream(), "UTF-8"));
String line;
while ((line = reader.readLine()) != null) {
// 读取处理逻辑
readLine(url, line);
}
} catch (Throwable t) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Failed to load extension of extensible " + interfaceName
+ " from classloader: " + classLoader + " and file:" + url, t);
}
} finally {
if (reader != null) {
reader.close();
}
}
}
}
}

protected void readLine(URL url, String line) {
// 解析一行，获取他们的别名和className，这个方法里对空做了处理，别名为空时仍可正常返回
String[] aliasAndClassName = parseAliasAndClassName(line);
if (aliasAndClassName == null || aliasAndClassName.length != 2) {
return;
}
String alias = aliasAndClassName[0];
String className = aliasAndClassName[1];
// 读取配置的实现类
Class tmp;
try {
tmp = ClassUtils.forName(className, false);
} catch (Throwable e) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Extension {} of extensible {} is disabled, cause by: {}",
className, interfaceName, ExceptionUtils.toShortString(e, 2));
}
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Extension " + className + " of extensible " + interfaceName + " is disabled.", e);
}
return;
}
// 用来判断A类是否是B类的子类或者子接口,Object是所有类的父类
// [isAssignableFrom](https://www.cnblogs.com/paul-lb/p/11344584.html)
if (!interfaceClass.isAssignableFrom(tmp)) {
throw new IllegalArgumentException("Error when load extension of extensible " + interfaceName +
" from file:" + url + ", " + className + " is not subtype of interface.");
}
Class implClass = (Class) tmp;
// 检查是否有可扩展标识，就是子类上面的Extension 注解
Extension extension = implClass.getAnnotation(Extension.class);
if (extension == null) {
throw new IllegalArgumentException("Error when load extension of extensible " + interfaceName +
" from file:" + url + ", " + className + " must add annotation @Extension.");
} else {
// 获取注解上面的扩展点名字必须写，不写就会报错
String aliasInCode = extension.value();
if (StringUtils.isBlank(aliasInCode)) {
// 扩展实现类未配置@Extension 标签
throw new IllegalArgumentException("Error when load extension of extensible " + interfaceClass +
" from file:" + url + ", " + className + "&＃39;s alias of @Extension is blank");
}
// 从配置文件取的别名，前面说了如果别名为空仍可正常返回
if (alias == null) {
// spi文件里没配置，用代码里的
alias = aliasInCode;
} else {
// spi文件里配置的和代码里的不一致
if (!aliasInCode.equals(alias)) {
throw new IllegalArgumentException("Error when load extension of extensible " + interfaceName +
" from file:" + url + ", aliases of " + className + " are " +
"not equal between " + aliasInCode + "(code) and " + alias + "(file).");
}
}
// 接口需要编号，实现类没设置
if (extensible.coded() && extension.code() <0) {
throw new IllegalArgumentException("Error when load extension of extensible " + interfaceName +
" from file:" + url + ", code of @Extension must >=0 at " + className + ".");
}
}
// 不可以是default和*
if (StringUtils.DEFAULT.equals(alias) || StringUtils.ALL.equals(alias)) {
throw new IllegalArgumentException("Error when load extension of extensible " + interfaceName +
" from file:" + url + ", alias of @Extension must not \"default\" and \"*\" at " + className + ".");
}
// 检查是否有存在同名的
ExtensionClass old = all.get(alias);
ExtensionClass extensiOnClass= null;
if (old != null) {
// 如果当前扩展可以覆盖其它同名扩展
if (extension.override()) {
// 如果优先级还没有旧的高，则忽略
if (extension.order() if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Extension of extensible {} with alias {} override from {} to {} failure, " +
"cause by: order of old extension is higher",
interfaceName, alias, old.getClazz(), implClass);
}
} else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Extension of extensible {} with alias {}: {} has been override to {}",
interfaceName, alias, old.getClazz(), implClass);
}
// 如果当前扩展可以覆盖其它同名扩展
extensiOnClass= buildClass(extension, implClass, alias);
}
}
// 如果旧扩展是可覆盖的
else {
if (old.isOverride() && old.getOrder() >= extension.order()) {
// 如果已加载覆盖扩展，再加载到原始扩展
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Extension of extensible {} with alias {}: {} has been loaded, ignore origin {}",
interfaceName, alias, old.getClazz(), implClass);
}
} else {
// 如果不能被覆盖，抛出已存在异常
throw new IllegalStateException(
"Error when load extension of extensible " + interfaceClass + " from file:" + url +
", Duplicate class with same alias: " + alias + ", " + old.getClazz() + " and " + implClass);
}
}
} else {
extensiOnClass= buildClass(extension, implClass, alias);
}
if (extensionClass != null) {
// 检查是否有互斥的扩展点
for (Map.Entry> entry : all.entrySet()) {
ExtensionClass existed = entry.getValue();
if (extensionClass.getOrder() >= existed.getOrder()) {
// 新的优先级 >= 老的优先级，检查新的扩展是否排除老的扩展
String[] rejection = extensionClass.getRejection();
if (CommonUtils.isNotEmpty(rejection)) {
for (String rej : rejection) {
existed = all.get(rej);
if (existed == null || extensionClass.getOrder() continue;
}
ExtensionClass removed = all.remove(rej);
if (removed != null) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info(
"Extension of extensible {} with alias {}: {} has been reject by new {}",
interfaceName, removed.getAlias(), removed.getClazz(), implClass);
}
}
}
}
} else {
String[] rejection = existed.getRejection();
if (CommonUtils.isNotEmpty(rejection)) {
for (String rej : rejection) {
if (rej.equals(extensionClass.getAlias())) {
// 被其它扩展排掉
if (LOGGER.isInfoEnabled()) {
LOGGER.info(
"Extension of extensible {} with alias {}: {} has been reject by old {}",
interfaceName, alias, implClass, existed.getClazz());
return;
}
}
}
}
}
}
loadSuccess(alias, extensionClass);
}
}

// 将实现类class别名等属性初始化到ExtensionClass里
private ExtensionClass buildClass(Extension extension, Class implClass, String alias) {
ExtensionClass extensiOnClass= new ExtensionClass(implClass, alias);
extensionClass.setCode(extension.code());
extensionClass.setSingleton(extensible.singleton());
extensionClass.setOrder(extension.order());
extensionClass.setOverride(extension.override());
extensionClass.setRejection(extension.rejection());
return extensionClass;
}
// 如果有监听器则通知监听器，加入全部的加载的实现类 {"alias":ExtensionClass} all
private void loadSuccess(String alias, ExtensionClass extensionClass) {
if (listener != null) {
try {
listener.onLoad(extensionClass); // 加载完毕，通知监听器
all.put(alias, extensionClass);
} catch (Exception e) {
LOGGER.error("Error when load extension of extensible " + interfaceClass + " with alias: "
+ alias + ".", e);
}
} else {
all.put(alias, extensionClass);
}
}

讲到这里我们spi的加载过程就讲完了，我们现在来看下他最后的操作，怎么取初始化获得加载的class对象呢？

/**
* 加载全部模块
*/
public static void installModules() {
ExtensionLoader loader = ExtensionLoaderFactory.getExtensionLoader(Module.class);
String moduleLoadList = RpcConfigs.getStringValue(RpcOptions.MODULE_LOAD_LIST);
for (Map.Entry> o : loader.getAllExtensions().entrySet()) {
String moduleName = o.getKey();
// 从这里可以看出来调用的是getExtInstance()
Module module = o.getValue().getExtInstance();
// judge need load from rpc option
if (needLoad(moduleLoadList, moduleName)) {
// judge need load from implement
if (module.needLoad()) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Install Module: {}", moduleName);
}
module.install();
INSTALLED_MODULES.put(moduleName, module);
} else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("The module " + moduleName + " does not need to be loaded.");
}
}
} else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("The module " + moduleName + " is not in the module load list.");
}
}
}
}


/**
* 得到服务端实例对象，如果是单例则返回单例对象，如果不是则返回新创建的实例对象
*
* @return 扩展点对象实例
*/
public T getExtInstance() {
return getExtInstance(null, null);
}

/**
* 得到服务端实例对象，如果是单例则返回单例对象，如果不是则返回新创建的实例对象
*
* @param argTypes 构造函数参数类型
* @param args 构造函数参数值
* @return 扩展点对象实例 ext instance
*/
public T getExtInstance(Class[] argTypes, Object[] args) {
if (clazz != null) {
try {
// 这个singleton是在加载spi的时候从extensible上面取出注入的哦
if (singleton) { // 如果是单例
if (instance == null) {
synchronized (this) {
if (instance == null) {
// 这里面是自动检测构造方法是否跟传的参数一直，一直则使用该构造器构造对象
instance = ClassUtils.newInstanceWithArgs(clazz, argTypes, args);
}
}
}
return instance; // 保留单例
} else {
// 这里面是自动检测构造方法是否跟传的参数一直，一直则使用该构造器构造对象
return ClassUtils.newInstanceWithArgs(clazz, argTypes, args);
}
} catch (Exception e) {
throw new SofaRpcRuntimeException("create " + clazz.getCanonicalName() + " instance error", e);
}
}
throw new SofaRpcRuntimeException("Class of ExtensionClass is null");
}

从上面的分析可以看出来对spi扩展类的初始化都是根据所传参数自动调用构造器去构造出来的对象，只是区分了单例和非单例的扩展类，单例使用了锁+双重检查的方式，保证只会初始化一次！

好了，今天我们所有的SOFA RPC SPI的分析到这里就结束了，看完是不是觉得还挺简单的，下面我来小结下全过程：

1、根据传入的扩展类class接口获得到类名，Extensible注解

2、根据Extensible是否填写file去配置的相对路径下加载相应的扩展文件，并获得一个length为2的数组

3、根据数组中配置的className去加载相应的class对象，并判断该对象是否是传入的class的子类

4、根据extension上的配置去检查同名覆盖、互斥扩展点等校验

5、将数据初始化到ExtensionClass上

6、将获得的别名和ExtensionClass存入ConcurrentMap中以备下次使用

总结结束，是不是很简单！