ExtensionLoader的使用
Dubbo中随处可见这样的代码：
ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
这段代码非常重要，了解它的实现是掌握dubbo源码的重中之重，否则很多地方的源码会不知其所以然，接下来以获取ThreadPool为例，讲解dubbo如何通过ExtensionLoader根据url获取具体的ThreadPool；
Provider&ThreadPool
Dubbo服务端接收到所有ChannelState类型的请求，都在com.alibaba.dubbo.remoting.transport.dispatcher.all.AllChannelHandler中处理，以Provider端处理接收到的请求为例，处理的代码如下：
public void received(Channel channel, Object message) throws RemotingException {
ExecutorService cexecutor = getExecutorService();
try {
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
} catch (Throwable t) {
throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
}
}
getExecutorService()就是获取连接池ExecutorService的；对应的就是父类WrappedChannelHandler中的申明
protected final ExecutorService executor；在new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)时，调用父类WrappedChannelHandler的构造方法时初始化，源码如下：
public WrappedChannelHandler(ChannelHandler handler, URL url) {
this.handler = handler;
this.url = url;
executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) {
componentKey = Constants.CONSUMER_SIDE;
}
DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
dataStore.put(componentKey, Integer.toString(url.getPort()), executor);
}
从这段代码可知，获取ExecutorService的代码就是ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);，接下来我们看看dubbo如何通过这一行代码获取具体的ThreadPool;
ExtensionLoader
这个类非常重要，几乎贯穿在dubbo中每一个模块，例如Protocol，LoadBalance等；因为dubbo采用SPI微内核实现方式，所以外部可以从容扩展自定义实现，例如自定义实现一种线程池AfeiThreadPool，而dubbo决定具体采用哪个实现类，都在这个类中决定，ExtensionLoader主要有两大功能：1、获取ExtensionLoader实例，2、获取Adaptive实例。调用方得到Adaptive实例后就能调用具体业务实现，例如FixedThreadPool，下面的分析以获取ThreadPool的实现为例；
1. 获取ExtensionLoader实例
首先根据ThreadPool得到ExtensionLoader，实现源码如下：
public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) {
if (type == null)
throw new IllegalArgumentException("Extension type == null");
// 获取ExtensionLoader时参数即ThreadPool必须是接口类型
if(!type.isInterface()) {
throw new IllegalArgumentException("Extension type(" + type + ") is not interface!");
}
// 获取ExtensionLoader时参数即ThreadPool必须有SPI注解，否则不可以扩展
if(!withExtensionAnnotation(type)) {
throw new IllegalArgumentException("Extension type(" + type +
") is not extension, because WITHOUT @" + SPI.class.getSimpleName() + " Annotation!");
}
// 将接口类型与其对应的ExtensionLoader本地缓存到`ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS`中
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;
}
2. 获取Adaptive实例
得到ExtensionLoader后，再得到具体的Adaptive实例；其中cachedAdaptiveInstance定义为：Holder<Object> cachedAdaptiveInstance，如果本地没有缓存实例，那么调用createAdaptiveExtension()获取，源码如下：
public T getAdaptiveExtension() {
// 先从本地缓存中查询能否得到Adaptive实例
Object instance = cachedAdaptiveInstance.get();
if (instance == null) {
// double-check方式保证线程安全（单实例的一种实现方式）
if(createAdaptiveInstanceError == null) {
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;
}
Hold申明如下，用volatile来保证数据的一致性：
/**
* Helper Class for hold a value.
*
* @author william.liangf
*/
public class Holder<T> {
private volatile T value;
public void set(T value) {
this.value = value;
}
public T get() {
return value;
}
}
如果本地还没有缓存实例，那么调用createAdaptiveExtension()获取Adaptive实例，源码如下:
private T createAdaptiveExtension() {
try {
return injectExtension((T) getAdaptiveExtensionClass().newInstance());
} catch (Exception e) {
throw new IllegalStateException("Can not create adaptive extenstion " + type + ", cause: " + e.getMessage(), e);
}
}
getAdaptiveExtensionClass()源码如下，先调用getExtensionClasses();，如果不能得到Adaptive类，那么调用createAdaptiveExtensionClass()获取Adaptive类；
private Class<?> getAdaptiveExtensionClass() {
getExtensionClasses();
if (cachedAdaptiveClass != null) {
return cachedAdaptiveClass;
}
return cachedAdaptiveClass = createAdaptiveExtensionClass();
}
接下来我们看getExtensionClasses()方法中如何得到Adaptive扩展类，其主要分为如下几个步骤：
- 得到ThreadPool接口上申明的SPI注解的值（"fixed"），就是默认名称cachedDefaultName ；
- 遍历类路径下三个目录（"META-INF/dubbo/internal"，"META-INF/dubbo/"，"META-INF/services/"），查找所有命名为type.getName()即的文件，得到文件中所有申明的实现集合；
- 查找有Adaptive注解的类，就是要找的Adaptive类；
- 如果没有Adaptive类，那么尝试能否通过Wrapper方式获取类，如果可以获取，那么缓存到
Set<Class<?>> cachedWrapperClasses;中；- 把文件中得到的所有类缓存到
Holder<Map<String, Class<?>>> cachedClasses中；
private Map<String, Class<?>> loadExtensionClasses() {
// 如果入参类型申明了SPI注解，那么SPI注解中的value就是默认实现名称，即cachedDefaultName （SPI注解中value的值只能有1个，超过1个会抛出异常）；
final SPI defaultAnnotation = type.getAnnotation(SPI.class);
if(defaultAnnotation != null) {
String value = defaultAnnotation.value();
if(value != null && (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];
}
}
// 遍历类路径下的（DUBBO_INTERNAL_DIRECTORY，DUBBO_DIRECTORY，SERVICES_DIRECTORY）三个目录，将符合type类型的所有实现存到Map中；
Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>();
loadFile(extensionClasses, DUBBO_INTERNAL_DIRECTORY);
loadFile(extensionClasses, DUBBO_DIRECTORY);
loadFile(extensionClasses, SERVICES_DIRECTORY);
return extensionClasses;
}
2.2 创建Adaptive扩展类
由于getExtensionClasses()中不能得到ThreadPool类型的Adaptive类（带有
Adaptive注解），所以需要通过createAdaptiveExtensionClass()创建Adaptive扩展类，由源码可知，其创建Adaptive扩展类分为如下几个步骤：
- 通过createAdaptiveExtensionClassCode()得到用于生成Adaptive扩展类的源代码code
- 得到类加载器ClassLoader
- 通过扩展机制得到编译器（默认用JavassistCompiler）将扩展类源代码code编译成字节码从而得到Adaptive类
private Class<?> createAdaptiveExtensionClass() {
String code = createAdaptiveExtensionClassCode();
ClassLoader classLoader = findClassLoader();
compiler =
return classLoader);
}
通过DEBUG方式得到ThreadPool的Adaptive扩展类的源代码如下，由下面的源码可知获取ThreadPool类型的Adaptive类的步骤：
- 判断请求URL中是否申明threadpool，如果没有申明threadpool，那么默认为fixed，即扩展名称；
- 根据扩展名称调用
ExtensionLoader.getExtension(extName)得到具体的实现类
从用于生成Adaptive扩展类的源码可以得知，URL中对threadpool申明的优先级要高于SPI注解申明；所以如果我们想修改默认的ThreadPool实现方式，可以通过URL中指定threadpool；
package
import
public class ThreadPool$Adpative implements {
public java.util.concurrent.Executor arg0) {
if (arg0 == null) throw new IllegalArgumentException("url == null");
url = arg0;
String extName = url.getParameter("threadpool", "fixed");
if(extName == null) throw new IllegalStateException("Fail to get name from url(" + url.toString() + ") use keys([threadpool])");
extension =
return extension.getExecutor(arg0);
}
}
创建Adaptive扩展类后，看.getExtension(String)方法如何得到具体的实现类，实现源码如下，即首先尝试从本地缓存cachedInstances中获取，如果获取不到，那么调用createExtension(name)创建扩展类实例并本地缓存；这里dubbo的作者又用到了double-check方式保证线程安全实现单实例；
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;
}
createExtension(String name)部分实现源码如下，由于前面已经调用了getExtensionClasses()得到所有ThreadPool类型并本地缓存在Holder<Map<String, Class<?>>> cachedClasses中，cachedClasses.get()得到保存的数据如下：
private T createExtension(String name) {
Class<?> clazz = getExtensionClasses().get(name);
if (clazz == null) {
throw findException(name);
}
... ...
}
配置ThreadPool
得到具体的ThreadPool实现后，dubbo根据url调整ThreadPoolExecutor，以FixedThreadPool为例（dubbo provider端默认采用的连接池处理方式），源码如下：
public class FixedThreadPool implements ThreadPool {
public Executor getExecutor(URL url) {
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new LinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}
}
- 如果url中配置了线程名称，那么进行设置，否则基于Constants.DEFAULT_THREAD_NAME得到默认线程名称;
- 如果url中配置了线程数量，那么进行设置，否则用默认线程数量：Constants.DEFAULT_THREADS，即默认200个线程；
- 如果url中配置了线程队列，那么进行设置，否则用默认线程队列：Constants.DEFAULT_QUEUES，即默认为0；
例如我们可以在provider.xml中对线程池做如下自定义：
<dubbo:protocol threadpool="fixed" threads="288" queues="10"/>