了解服務(wù)發(fā)布
Dubbo官方文檔說明了服務(wù)提供者暴露服務(wù)的主過程，如圖所示：
首先ServiceConfig類拿到對(duì)外提供服務(wù)的實(shí)際類ref(如：HelloWorldImpl),然后通過ProxyFactory類的 getInvoker方法使用ref生成一個(gè)AbstractProxyInvoker實(shí)例割卖，到這一步就完成具體服務(wù)到Invoker的轉(zhuǎn)化裹赴。接下來就是Invoker轉(zhuǎn)換到Exporter的過程盅藻。Dubbo 處理服務(wù)暴露的關(guān)鍵就在Invoker轉(zhuǎn)換到Exporter的過程癞己，上圖中的紅色部分淘邻。
源碼分析
入口分析——ServiceBean
服務(wù)發(fā)布在spring的配置文件中配置如下：
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://code.alibabatech.com/schema/dubbo
http://code.alibabatech.com/schema/dubbo/dubbo.xsd">
<bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl" />
<dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService" />
</beans>
如果熟悉Spring自定義標(biāo)簽窑眯，應(yīng)該知道<dubbo:service>標(biāo)簽會(huì)轉(zhuǎn)換成ServiceBean屏积。通過DubboNamespaceHandler.init()方法可知，其源碼如下：
registerBeanDefinitionParser("service",
new DubboBeanDefinitionParser(ServiceBean.class, true));
ServiceBean類結(jié)構(gòu)如下圖所示：
dubbo暴露服務(wù)有兩種情況磅甩，一種是設(shè)置了延遲暴露（比如delay=”5000”）炊林，另外一種是沒有設(shè)置延遲暴露或者延遲設(shè)置為-1（delay=”-1”）：
1、設(shè)置了延遲暴露卷要，dubbo在Spring實(shí)例化bean（initializeBean）的時(shí)候會(huì)對(duì)實(shí)現(xiàn)了InitializingBean的類進(jìn)行回調(diào)渣聚，回調(diào)方法是afterPropertySet()独榴，如果設(shè)置了延遲暴露，dubbo在這個(gè)方法中進(jìn)行服務(wù)的發(fā)布奕枝。
2棺榔、沒有設(shè)置延遲或者延遲為-1，dubbo會(huì)在Spring實(shí)例化完bean之后隘道，在刷新容器最后一步發(fā)布ContextRefreshEvent事件的時(shí)候症歇，通知實(shí)現(xiàn)了ApplicationListener的類進(jìn)行回調(diào)onApplicationEvent，dubbo會(huì)在這個(gè)方法中發(fā)布服務(wù)薄声。
但是不管延遲與否当船，都是使用ServiceConfig的export()方法進(jìn)行服務(wù)的暴露。使用export初始化的時(shí)候會(huì)將Bean對(duì)象轉(zhuǎn)換成URL格式默辨，所有Bean屬性轉(zhuǎn)換成URL的參數(shù)德频。
ServiceConfig.export()方法
ServiceConfig.export()的流程圖如下：
1、export()方法先判斷是否需要延遲暴露缩幸，如果設(shè)置了延遲壹置，則通過一個(gè)后臺(tái)線程調(diào)用doExport()方法；反之表谊，直接調(diào)用doExport()方法钞护。
2、doExport方法先執(zhí)行一系列的檢查方法爆办，然后調(diào)用doExportUrls方法难咕。檢查方法會(huì)檢測(cè)dubbo的配置是否在Spring配置文件中聲明，沒有的話讀取properties文件初始化距辆。
3余佃、doExportUrls方法先調(diào)用loadRegistries獲取所有的注冊(cè)中心url，然后遍歷調(diào)用doExportUrlsFor1Protocol方法跨算。
4爆土、doExportUrlsFor1Protocol()先將Bean屬性轉(zhuǎn)換成URL對(duì)象，然后根據(jù)不同協(xié)議將服務(wù)已URL形式發(fā)布诸蚕。如果scope配置為none則不暴露步势，如果服務(wù)未配置成remote，則本地暴露exportLocal背犯，如果未配置成local坏瘩，則遠(yuǎn)程暴露。
暴露服務(wù)的核心是由doExportUrlsFor1Protocol()方法處理的漠魏，其源碼如下：
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
// 1桑腮、Bean屬性轉(zhuǎn)換URL
URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.hasExtension(url.getProtocol())) {
url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.getExtension(url.getProtocol()).getConfigurator(url).configure(url);
}
String scope = url.getParameter(Constants.SCOPE_KEY);
if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
// scope配置不是remote的情況下做本地暴露
if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
exportLocal(url);
}
// scope配置不是local則暴露為遠(yuǎn)程服務(wù)
if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
if (registryURLs != null && registryURLs.size() > 0
&& url.getParameter("register", true)) {
for (URL registryURL : registryURLs) {
url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
URL monitorUrl = loadMonitor(registryURL);
if (monitorUrl != null) {
url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
}
// 2、具體服務(wù)到invoker的轉(zhuǎn)換
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
// 3蛉幸、invoker轉(zhuǎn)換為exporter
Exporter<?> exporter = protocol.export(invoker);
exporters.add(exporter);
}
} else {
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
Exporter<?> exporter = protocol.export(invoker);
exporters.add(exporter);
}
}
}
this.urls.add(url);
}
具體服務(wù)到Invoker轉(zhuǎn)換
在分析具體服務(wù)到Invoker之前破讨，先來看看ServiceConfig的proxyFactory實(shí)例丛晦。其定義如下：
private static final ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
ProxyFactory的自適應(yīng)擴(kuò)展點(diǎn)的getInvoker方法源碼如下：
public com.alibaba.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, com.alibaba.dubbo.common.URL arg2) throws RpcException {
if (arg2 == null) throw new IllegalArgumentException("url == null");
com.alibaba.dubbo.common.URL url = arg2;
// 如果url沒有proxy參數(shù)，默認(rèn)為javassist
String extName = url.getParameter("proxy", "javassist");
if (extName == null)
throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
return extension.getInvoker(arg0, arg1, arg2);
}
通過上述源碼我們知道提陶，最終獲取invoker是由JavassistProxyFactory.getInvoker方法實(shí)現(xiàn)的烫沙，其源碼如下：
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// 獲取具體服務(wù)類的包裝對(duì)象
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
// 返回構(gòu)造invoker實(shí)例
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
};
}
流程總結(jié)：
1、根據(jù)url的proxy參數(shù)獲取對(duì)應(yīng)ProxyFactory類隙笆，默認(rèn)為JavassistProxyFactory锌蓄；
2、獲取具體服務(wù)類的包裝對(duì)象撑柔；
3瘸爽、返回構(gòu)造invoker實(shí)例，將服務(wù)具體類的方法調(diào)用封裝在doInvoke()方法中铅忿。
遠(yuǎn)程暴露
通過debug源碼剪决，protocol.export(invoker)的時(shí)序圖如下：
流程總結(jié)
1、構(gòu)建invoker過濾鏈檀训；
2柑潦、首先DubboProcotol的export方法將invoker轉(zhuǎn)換成DubboExporter，啟動(dòng)Server服務(wù)峻凫，然后將DubboExporter包裝為L(zhǎng)istenerExporterWrapper渗鬼。
3、RegistryProtocol的export方法向注冊(cè)中心注冊(cè)服務(wù)提供者url和訂閱url荧琼，再次將ListenerExporterWrapper包裝為Exporter譬胎，覆蓋unexport()方法。
RegistryProtocol.export()方法的注冊(cè)和訂閱
下面我們分析下RegistryProtocol.export()方法的注冊(cè)和訂閱命锄，其源碼如下：
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數(shù)據(jù)
// provider://172.16.6.216:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&category=configurators&check=false&dubbo=2.5.3&interface=com.alibaba.dubbo.demo.DemoService&loadbalance=roundrobin&methods=sayHello&owner=william&pid=2154&side=provider×tamp=1520235259270
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
//保證每次export都返回一個(gè)新的exporter實(shí)例
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);
}
}
};
}
注冊(cè)
由于使用zookeeper作為注冊(cè)中心堰乔，所以registry為ZookeeperRegistry。ZookeeperRegistry.registry過程如下：
ZookeeperRegistry.doRegister方法如下：
protected void doRegister(URL url) {
try {
zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
doRegister方法將url轉(zhuǎn)換為注冊(cè)zookeeper的path累舷，創(chuàng)建臨時(shí)節(jié)點(diǎn)。 臨時(shí)節(jié)點(diǎn)在zookeeper會(huì)話失效后會(huì)自動(dòng)刪除的夹孔，Dubbo如何解決這個(gè)問題被盈。
public FailbackRegistry(URL url) {
super(url);
int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
public void run() {
// 檢測(cè)并連接注冊(cè)中心
try {
retry();
} catch (Throwable t) { // 防御性容錯(cuò)
logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
}
}
}, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
}
訂閱
protected void doSubscribe(final URL url, final NotifyListener listener) {
try {
if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
String root = toRootPath();
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
if (listeners == null) {
zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
listeners = zkListeners.get(url);
}
ChildListener zkListener = listeners.get(listener);
if (zkListener == null) {
listeners.putIfAbsent(listener, new ChildListener() {
public void childChanged(String parentPath, List<String> currentChilds) {
for (String child : currentChilds) {
if (! anyServices.contains(child)) {
anyServices.add(child);
subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child,
Constants.CHECK_KEY, String.valueOf(false)), listener);
}
}
}
});
zkListener = listeners.get(listener);
}
zkClient.create(root, false);
List<String> services = zkClient.addChildListener(root, zkListener);
if (services != null && services.size() > 0) {
anyServices.addAll(services);
for (String service : services) {
subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
Constants.CHECK_KEY, String.valueOf(false)), listener);
}
}
} else {
List<URL> urls = new ArrayList<URL>();
// /dubbo/com.alibaba.dubbo.demo.DemoService/configurators
for (String path : toCategoriesPath(url)) {
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
if (listeners == null) {
zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
listeners = zkListeners.get(url);
}
ChildListener zkListener = listeners.get(listener);
if (zkListener == null) {
listeners.putIfAbsent(listener, new ChildListener() {
public void childChanged(String parentPath, List<String> currentChilds) {
//
ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
}
});
zkListener = listeners.get(listener);
}
zkClient.create(path, false);
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
notify(url, listener, urls);
}
} catch (Throwable e) {
throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
doSubscribe方法首先將NotifyListener轉(zhuǎn)換為Zookeeper Listener，創(chuàng)建/dubbo/com.alibaba.dubbo.demo.DemoService/configurators持久節(jié)點(diǎn)并訂閱搭伤。
為什么要訂閱/dubbo/com.alibaba.dubbo.demo.DemoService/configurators這個(gè)節(jié)點(diǎn)只怎。
