dubbo源码阅读
2018-11-22 本文已影响26人
任嘉平生愿
ExtensionLoader
将每个方法标为（1）（2）（3）（4）
这个类可是dubbo的精髓在下面这个方法上打上断点 （2）
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!");
}
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;
}
第一次执行得到的
loaderinterface com.alibaba.dubbo.common.extension.ExtensionFactory
上面的type判断是不是接口和是不是有@spi注解侧面说明ExtensionLoader就是对@spi类的扩展和实现
1.png
继续走返回到构造函数(1)
private ExtensionLoader(Class<?> type) {
this.type = type;
objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension());
}
接着进入适配器扩展getAdaptiveExtension (3)
public T getAdaptiveExtension() {
Object instance = cachedAdaptiveInstance.get();//
if (instance == null) {
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;
}
这个方法中用个 instance = createAdaptiveExtension();
createAdaptiveExtension (4)
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);
}
}
image.png
可以看到instance中有一个factory下面有两个Factory一个spring一个spi
(这里我有一个猜想：这里的适配器作用可能是适配spring和spi)
ExtensionLoader（1）构造函数走完就做了一件事得到了大Factory
继续
第二次走getExtensionLoader（2）
image.png
这次的type变成了interface com.alibaba.dubbo.rpc.Protocol
loader里面
之后几次都是一样的过程直到这个方法的调用createAdaptiveExtensionClass(5)
private Class<?> createAdaptiveExtensionClass() {
String code = createAdaptiveExtensionClassCode();
ClassLoader classLoader = findClassLoader();
com.alibaba.dubbo.common.compiler.Compiler compiler = ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.common.compiler.Compiler.class).getAdaptiveExtension();
return compiler.compile(code, classLoader);
}
这个方法名可知就是创建适配扩展类
这句代码就很精髓了
image.png
可以看到得到了一个code和一个classLoader
因为classLoad有$这个符号是不能查看的但是可以猜想到是Protocol这个类
为什么呢？我把code里面的字符串贴出来就明白了
package com.alibaba.dubbo.rpc;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class Protocol$Adaptive implements com.alibaba.dubbo.rpc.Protocol {
public void destroy() {throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
}
public int getDefaultPort() {throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
}
public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException {
if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
return extension.export(arg0);
}
public com.alibaba.dubbo.rpc.Invoker refer(java.lang.Class arg0, com.alibaba.dubbo.common.URL arg1) throws com.alibaba.dubbo.rpc.RpcException {
if (arg1 == null) throw new IllegalArgumentException("url == null");
com.alibaba.dubbo.common.URL url = arg1;
String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
return extension.refer(arg0, arg1);
}
}
看到这个code你会发现很神奇！code里居然塞了一个Protocol$Adaptive这个类还实现了Protocol接口！
得出一个结论这个ExtensionLoader就是在制造Protocol$Adaptive这玩意
再来看一下Protocol 接口 一共4个方法
@SPI("dubbo")
public interface Protocol {
/**
* 获取缺省端口，当用户没有配置端口时使用。
*
* @return 缺省端口
*/
int getDefaultPort();
/**
* 暴露远程服务：<br>
* 1. 协议在接收请求时，应记录请求来源方地址信息：RpcContext.getContext().setRemoteAddress();<br>
* 2. export()必须是幂等的，也就是暴露同一个URL的Invoker两次，和暴露一次没有区别。<br>
* 3. export()传入的Invoker由框架实现并传入，协议不需要关心。<br>
*
* @param <T> 服务的类型
* @param invoker 服务的执行体
* @return exporter 暴露服务的引用，用于取消暴露
* @throws RpcException 当暴露服务出错时抛出，比如端口已占用
*/
@Adaptive
<T> Exporter<T> export(Invoker<T> invoker) throws RpcException;
/**
* 引用远程服务：<br>
* 1. 当用户调用refer()所返回的Invoker对象的invoke()方法时，协议需相应执行同URL远端export()传入的Invoker对象的invoke()方法。<br>
* 2. refer()返回的Invoker由协议实现，协议通常需要在此Invoker中发送远程请求。<br>
* 3. 当url中有设置check=false时，连接失败不能抛出异常，并内部自动恢复。<br>
*
* @param <T> 服务的类型
* @param type 服务的类型
* @param url 远程服务的URL地址
* @return invoker 服务的本地代理
* @throws RpcException 当连接服务提供方失败时抛出
*/
@Adaptive
<T> Invoker<T> refer(Class<T> type, URL url) throws RpcException;
/**
* 释放协议：<br>
* 1. 取消该协议所有已经暴露和引用的服务。<br>
* 2. 释放协议所占用的所有资源，比如连接和端口。<br>
* 3. 协议在释放后，依然能暴露和引用新的服务。<br>
*/
void destroy();
}
发现了什么！~~好吧有@Adaptive这个注解
而对照生成的Protocol$Adaptive
你会发现他实际就实现了这两个方法Invoker（）和export（）
我们再来观察一下Protocol的另一个实现类RegistryProtocol
package com.enjoy.test;
/**
* RegistryProtocol
*
* @author william.liangf
* @author chao.liuc
*/
public class RegistryProtocol implements Protocol {
private final static Logger logger = LoggerFactory.getLogger(RegistryProtocol.class);
private static RegistryProtocol INSTANCE;
private final Map<URL, NotifyListener> overrideListeners = new ConcurrentHashMap<URL, NotifyListener>();
//用于解决rmi重复暴露端口冲突的问题，已经暴露过的服务不再重新暴露
//providerurl <--> exporter
private final Map<String, ExporterChangeableWrapper<?>> bounds = new ConcurrentHashMap<String, ExporterChangeableWrapper<?>>();
private Cluster cluster;
private Protocol protocol;
private RegistryFactory registryFactory;
private ProxyFactory proxyFactory;
public RegistryProtocol() {
INSTANCE = this;
}
public static RegistryProtocol getRegistryProtocol() {
if (INSTANCE == null) {
ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(Constants.REGISTRY_PROTOCOL); // load
}
return INSTANCE;
}
//过滤URL中不需要输出的参数(以点号开头的)
private static String[] getFilteredKeys(URL url) {
Map<String, String> params = url.getParameters();
if (params != null && !params.isEmpty()) {
List<String> filteredKeys = new ArrayList<String>();
for (Map.Entry<String, String> entry : params.entrySet()) {
if (entry != null && entry.getKey() != null && entry.getKey().startsWith(Constants.HIDE_KEY_PREFIX)) {
filteredKeys.add(entry.getKey());
}
}
return filteredKeys.toArray(new String[filteredKeys.size()]);
} else {
return new String[]{};
}
}
public void setCluster(Cluster cluster) {
this.cluster = cluster;
}
public void setProtocol(Protocol protocol) {
this.protocol = protocol;
}
public void setRegistryFactory(RegistryFactory registryFactory) {
this.registryFactory = registryFactory;
}
public void setProxyFactory(ProxyFactory proxyFactory) {
this.proxyFactory = proxyFactory;
}
public int getDefaultPort() {
return 9090;
}
public Map<URL, NotifyListener> getOverrideListeners() {
return overrideListeners;
}
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
//export invoker
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
//registry provider
final Registry registry = getRegistry(originInvoker);
final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
registry.register(registedProviderUrl);
// 订阅override数据
// FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
//保证每次export都返回一个新的exporter实例
return new Exporter<T>() {
public Invoker<T> getInvoker() {
return exporter.getInvoker();
}
public void unexport() {
try {
exporter.unexport();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
try {
registry.unregister(registedProviderUrl);
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
try {
overrideListeners.remove(overrideSubscribeUrl);
registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
};
}
@SuppressWarnings("unchecked")
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
String key = getCacheKey(originInvoker);
ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
synchronized (bounds) {
exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
bounds.put(key, exporter);
}
}
}
return exporter;
}
/**
* 对修改了url的invoker重新export
*
* @param originInvoker
* @param newInvokerUrl
*/
@SuppressWarnings("unchecked")
private <T> void doChangeLocalExport(final Invoker<T> originInvoker, URL newInvokerUrl) {
String key = getCacheKey(originInvoker);
final ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
logger.warn(new IllegalStateException("error state, exporter should not be null"));
} else {
final Invoker<T> invokerDelegete = new InvokerDelegete<T>(originInvoker, newInvokerUrl);
exporter.setExporter(protocol.export(invokerDelegete));
}
}
/**
* 根据invoker的地址获取registry实例
*
* @param originInvoker
* @return
*/
private Registry getRegistry(final Invoker<?> originInvoker) {
URL registryUrl = originInvoker.getUrl();
if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
}
return registryFactory.getRegistry(registryUrl);
}
/**
* 返回注册到注册中心的URL，对URL参数进行一次过滤
*
* @param originInvoker
* @return
*/
private URL getRegistedProviderUrl(final Invoker<?> originInvoker) {
URL providerUrl = getProviderUrl(originInvoker);
//注册中心看到的地址
final URL registedProviderUrl = providerUrl.removeParameters(getFilteredKeys(providerUrl))
.removeParameter(Constants.MONITOR_KEY)
.removeParameter(Constants.BIND_IP_KEY)
.removeParameter(Constants.BIND_PORT_KEY);
return registedProviderUrl;
}
private URL getSubscribedOverrideUrl(URL registedProviderUrl) {
return registedProviderUrl.setProtocol(Constants.PROVIDER_PROTOCOL)
.addParameters(Constants.CATEGORY_KEY, Constants.CONFIGURATORS_CATEGORY,
Constants.CHECK_KEY, String.valueOf(false));
}
/**
* 通过invoker的url 获取 providerUrl的地址
*
* @param origininvoker
* @return
*/
private URL getProviderUrl(final Invoker<?> origininvoker) {
String export = origininvoker.getUrl().getParameterAndDecoded(Constants.EXPORT_KEY);
if (export == null || export.length() == 0) {
throw new IllegalArgumentException("The registry export url is null! registry: " + origininvoker.getUrl());
}
URL providerUrl = URL.valueOf(export);
return providerUrl;
}
/**
* 获取invoker在bounds中缓存的key
*
* @param originInvoker
* @return
*/
private String getCacheKey(final Invoker<?> originInvoker) {
URL providerUrl = getProviderUrl(originInvoker);
String key = providerUrl.removeParameters("dynamic", "enabled").toFullString();
return key;
}
@SuppressWarnings("unchecked")
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
String group = qs.get(Constants.GROUP_KEY);
if (group != null && group.length() > 0) {
if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
|| "*".equals(group)) {
return doRefer(getMergeableCluster(), registry, type, url);
}
}
return doRefer(cluster, registry, type, url);
}
private Cluster getMergeableCluster() {
return ExtensionLoader.getExtensionLoader(Cluster.class).getExtension("mergeable");
}
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
directory.setRegistry(registry);
directory.setProtocol(protocol);
// REFER_KEY的所有属性
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
&& url.getParameter(Constants.REGISTER_KEY, true)) {
registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
Constants.CHECK_KEY, String.valueOf(false)));
}
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
return cluster.join(directory);
}
public void destroy() {
List<Exporter<?>> exporters = new ArrayList<Exporter<?>>(bounds.values());
for (Exporter<?> exporter : exporters) {
exporter.unexport();
}
bounds.clear();
}
public static class InvokerDelegete<T> extends InvokerWrapper<T> {
private final Invoker<T> invoker;
/**
* @param invoker
* @param url invoker.getUrl返回此值
*/
public InvokerDelegete(Invoker<T> invoker, URL url) {
super(invoker, url);
this.invoker = invoker;
}
public Invoker<T> getInvoker() {
if (invoker instanceof InvokerDelegete) {
return ((InvokerDelegete<T>) invoker).getInvoker();
} else {
return invoker;
}
}
}
/*重新export 1.protocol中的exporter destory问题
*1.要求registryprotocol返回的exporter可以正常destroy
*2.notify后不需要重新向注册中心注册
*3.export 方法传入的invoker最好能一直作为exporter的invoker.
*/
private class OverrideListener implements NotifyListener {
private final URL subscribeUrl;
private final Invoker originInvoker;
public OverrideListener(URL subscribeUrl, Invoker originalInvoker) {
this.subscribeUrl = subscribeUrl;
this.originInvoker = originalInvoker;
}
/**
* @param urls 已注册信息列表，总不为空，含义同{@link RegistryService#lookup(URL)}的返回值。
*/
public synchronized void notify(List<URL> urls) {
logger.debug("original override urls: " + urls);
List<URL> matchedUrls = getMatchedUrls(urls, subscribeUrl);
logger.debug("subscribe url: " + subscribeUrl + ", override urls: " + matchedUrls);
//没有匹配的
if (matchedUrls.isEmpty()) {
return;
}
List<Configurator> configurators = RegistryDirectory.toConfigurators(matchedUrls);
final Invoker<?> invoker;
if (originInvoker instanceof InvokerDelegete) {
invoker = ((InvokerDelegete<?>) originInvoker).getInvoker();
} else {
invoker = originInvoker;
}
//最原始的invoker
URL originUrl = RegistryProtocol.this.getProviderUrl(invoker);
String key = getCacheKey(originInvoker);
ExporterChangeableWrapper<?> exporter = bounds.get(key);
if (exporter == null) {
logger.warn(new IllegalStateException("error state, exporter should not be null"));
return;
}
//当前的，可能经过了多次merge
URL currentUrl = exporter.getInvoker().getUrl();
//与本次配置merge的
URL newUrl = getConfigedInvokerUrl(configurators, originUrl);
if (!currentUrl.equals(newUrl)) {
RegistryProtocol.this.doChangeLocalExport(originInvoker, newUrl);
logger.info("exported provider url changed, origin url: " + originUrl + ", old export url: " + currentUrl + ", new export url: " + newUrl);
}
}
private List<URL> getMatchedUrls(List<URL> configuratorUrls, URL currentSubscribe) {
List<URL> result = new ArrayList<URL>();
for (URL url : configuratorUrls) {
URL overrideUrl = url;
// 兼容旧版本
if (url.getParameter(Constants.CATEGORY_KEY) == null
&& Constants.OVERRIDE_PROTOCOL.equals(url.getProtocol())) {
overrideUrl = url.addParameter(Constants.CATEGORY_KEY, Constants.CONFIGURATORS_CATEGORY);
}
//检查是不是要应用到当前服务上
if (UrlUtils.isMatch(currentSubscribe, overrideUrl)) {
result.add(url);
}
}
return result;
}
//合并配置的url
private URL getConfigedInvokerUrl(List<Configurator> configurators, URL url) {
for (Configurator configurator : configurators) {
url = configurator.configure(url);
}
return url;
}
}
/**
* exporter代理,建立返回的exporter与protocol export出的exporter的对应关系，在override时可以进行关系修改.
*
* @param <T>
* @author chao.liuc
*/
private class ExporterChangeableWrapper<T> implements Exporter<T> {
private final Invoker<T> originInvoker;
private Exporter<T> exporter;
public ExporterChangeableWrapper(Exporter<T> exporter, Invoker<T> originInvoker) {
this.exporter = exporter;
this.originInvoker = originInvoker;
}
public Invoker<T> getOriginInvoker() {
return originInvoker;
}
public Invoker<T> getInvoker() {
return exporter.getInvoker();
}
public void setExporter(Exporter<T> exporter) {
this.exporter = exporter;
}
public void unexport() {
String key = getCacheKey(this.originInvoker);
bounds.remove(key);
exporter.unexport();
}
}
}
粗略看完这个类你会发现他除了实现那两个方法之外还有其他的
这又说明了一个问题dubbo暴露的服务其实在做事的就是两个方法！
就是服务端通过refer()和消费端通过export（）！！！
回过头我们再观察code生成的代码
image.png
发现一个问题refer()和export（）中都是同样做了一件事就是得到URL
url是什么我们不知道因为不能debug但是我们接着往下看
继续
然后一直往下走到了这个类ReferenceConfig
中的这个方法
private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
得到一个结论：妈的绕了一大圈就是得到了一个代理的refprotocol
继续
一直在断点getExtensionLoader
会发现type一直在变规律就是先出来一个
interface com.alibaba.dubbo.common.extension.ExtensionFactory
再出来一个带有@SPI注解的类目的只有一个就是得到refXXX
而且一直继续的情况下你会发现一个神奇的事情就是又很多的@spi类接口都没实现了
继续...
突然就结束了！至此debug结束了！？
那么问题来了我们前面得到了refXXX和一个不知道的URL到底是干啥的？
好吧那我们在顺藤摸瓜之前有debug到一个AbastractReferenceConfig中找到实现类ReferenceBean
image.png
这个联系也不是多端的因为在配置dubbo的配置文件的时候
<dubbo:reference id="userService" interface="com.enjoy.service.UserService" protocol="dubbo" loadbalance="first" />
<dubbo:reference id="orderService" interface="com.enjoy.service.OrderService" loadbalance="last" >
<dubbo:method name="submit" async="true" onreturn="callBack.onOrderSubmit" onthrow="callBack.onError"/>
<dubbo:method name="cancel" async="true" />
</dubbo:reference>
但是走到这里又不对因为到这里其实配置一直结束了url也是个中间结果
所以回过头继续寻找终于还是在ExtensionLoader类中找到了答案！
到这里在this中找ExtensionLoader类中的数据可以得到！
image.png
他是将原来定义好的protocal全部加载出来然后很具我们配置的名词去调用实现类比如protocal中的dubbo和rmi根据匹配然后找到实现了类去继承
至此dubbo的ExtensionLoader类结束
