RPC框架（六）dubbo原始碼分析--dubbo服務提供者初始化
一、概述
dubbo服務提供者由dubbo:service來定義，從前面可以看到，Spring把dubbo:service解析成一個ServiceBean，ServiceBean實現了ApplicationListener和InitializingBean介面，ServiceBean有個核心方法export，在這個方法中初始化服務提供者並且暴露遠端服務。這個方法在bean初始化或容器中所有bean重新整理完畢時被呼叫，根據provider的延遲設定決定，如果設定了延遲（delay屬性）在bean初始化結束之後呼叫否則容易重新整理事件中被呼叫，預設會延遲export，即在所有bean的重新整理結束時間中被呼叫。
在com.alibaba.dubbo.config.spring.ServiceBean類下
public void onApplicationEvent(ApplicationEvent event) {
if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
if (isDelay() && ! isExported() && ! isUnexported()) {
if (logger.isInfoEnabled()) {
logger.info("The service ready on spring started. service: " 在export方法中，總體上export方法暴露服務時主要做了下列這些步驟：
- 選用服務埠
- 生成URL物件
- 生成本地服務代理（如果需要）
- 生成遠端服務代理
- 啟用服務監聽
- 服務註冊到註冊中心
二、選用服務埠
服務需要一個埠來進行服務呼叫偵聽，框架預設會從20880開始選定一個未被佔用的埠來提供服務，也可以在配置中的port引數指定服務的埠：
com.alibaba.dubbo.config.ServiceConfig.doExportUrlsFor1Protocol(ProtocolConfig, List)
final int defaultPort = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(name).getDefaultPort();
if (port == null || port == 0) {
port = defaultPort;
}
if (port == null || port <= 0) {
port = getRandomPort(name);
if (port == null || port < 0) {
port = NetUtils.getAvailablePort(defaultPort);
putRandomPort(name, port);
}
logger.warn("Use random available port(" + port + ") for protocol " + name);
}三、生成URL物件
URL物件是dubbo框架中的核心模型，它攜帶了呼叫鏈路中註冊中心、消費端配置、提供端配置等全部資訊。無論是消費端代理還是提供端代理都需持有一個URL物件攜帶服務呼叫時的上下文資訊。
在服務初始化時需要對註冊中心進行訪問，所以需要一個URL來跟註冊中心通訊，根據應用的註冊中心配置來生成這個URL，URL協議是registry，包含了註冊中心地址、埠，path部分是com.alibaba.dubbo.registry.RegistryService類名，其中還有dubbo版本、應用名稱等資訊放在URL的引數中，格式化之後是這種形式：registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=…
除了註冊中心URL，還需要生成一個URL來攜帶服務本身的資訊，協議由應用側指定，在我們的示例配置中是dubbo協議，host是本機ip，埠是上面選中的隨機或配置中指定的埠，path是服務對應的介面class全名（含包路徑），新增side引數provider，dubbo版本號、服務方法名稱、group等等，格式化之後是這種形式：
dubbo://127.0.0.1:20880/com.netease.haitao.mykaola.generic.api.InternalStaffServiceFacade?…&createInternalStaffOperDetail.timeout=10000&…&default.group=hzchenyunyun1&…&dubbo=3.0.0&…&methods=…&…&side=provider&… 這個url會被設定到registry URL的export屬性中，這點很重要，後面的初始化過程是圍繞registry URL，需要從這個export屬性中拿到提供者服務URL。
四、生成本地服務代理
如果scope屬性沒有被設定成remote，服務同時會在本地暴露，生成一個本地服務代理物件，這裡會生成一個新的URL，協議是代表本地服務的injvm，host是127.0.0.1埠是0，生成一個InjvmExporter，這時本地呼叫dubbo介面時直接呼叫本地代理不走網路請求。生成Exporter的過程和生成遠端Exporter的過程類似，在後面詳細描述。
com.alibaba.dubbo.config.ServiceConfig.exportLocal(URL)
private void exportLocal(URL url) {
if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
URL local = URL.valueOf(url.toFullString())
.setProtocol(Constants.LOCAL_PROTOCOL)
.setHost(NetUtils.LOCALHOST)
.setPort(0);
Exporter<?> exporter = protocol.export(
proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
exporters.add(exporter);
logger.info("Export dubbo service " + interfaceClass.getName() +" to local registry");
}
}五、生成遠端服務代理
如果scope屬性沒有被設定成local，生成遠端服務代理，框架提供了ProxyFactory生成服務代理（dubbo提供JDK動態代理和Javaassist代理，預設使用Javaassist，使用者也可以替換成其他的代理，這裡不展開），它會生成一個Invoker，該Invoker是服務ref的一個代理包含了攜帶服務資訊的URL物件。這裡的ref就是在dubbo:service中配置的ref屬性，指定服務的具體實現類，Invoker的invoke方法被呼叫時，最終會呼叫到ref指定的服務實現。這裡的ProxyFactory使用ExtensionLoader的自適應擴充套件點載入，如應用側沒有特別指定，預設的是JavassistProxyFactory。
com.alibaba.dubbo.config.ServiceConfig.doExportUrlsFor1Protocol(ProtocolConfig, List)
//如果配置不是local則暴露為遠端服務.(配置為local，則表示只暴露遠端服務)
if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
if (logger.isInfoEnabled()) {
logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
}
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());
}
if (logger.isInfoEnabled()) {
logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
}
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
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);
}
}public class JavassistProxyFactory extends AbstractProxyFactory {
@SuppressWarnings("unchecked")
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// TODO Wrapper類不能正確處理帶$的類名
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
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);
}
};
}
}從上面的程式碼裡面看到，建立Invoker之後接下來就是建立一個Exporter，由Protocol元件來建立。這裡的Protocol擴充套件點也是自適應載入，而當前的Url的協議是registry，自適應protocol最終會呼叫ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(“registry”)， Protocol定義了ProtocolListenerWrapper和ListenerExporterWrapper兩個Wrapper擴充套件點，而當前的Url的協議是registry，根據ExtensionLoader的載入規則，它會返回ProtocolListenerWrapper->ListenerExporterWrapper->RegistryProtocol物件鏈，對於registry協議，兩個Wrapper都不會做任何處理，會直接調到RegistryProtocol.export方法。
com.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
return protocol.export(invoker);
}
return new ListenerExporterWrapper<T>(protocol.export(invoker),
Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
.getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));
}com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
return protocol.export(invoker);
}
return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
}RegistryProtocol.export會呼叫doLocalExport方法，執行服務暴露邏輯：
com.alibaba.dubbo.registry.integration.RegistryProtocol.doLocalExport(Invoker)
@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 (ExporterChangeableWrapper<T>) exporter;
}接著呼叫protocol.export，這時候url發生了變化不再是registryUrl了，而是存放在registryUrl的export引數中的providerUrl，根據不同協議會定位到不同的擴充套件，示例配置是dubbo協議，對應的實現是DubboProtocol，同樣地根據Wrapper擴充套件點載入機制會加載出ProtocolListenerWrapper和 ListenerExporterWrapper兩個Wrapper，然後依次呼叫ProtocolListenerWrapper->ListenerExporterWrapper->DubboProtocol的export方法。從上面的程式碼可以看到，ProtocolListenerWrapper.export會建立一個ListenerExporterWrapper例項，並新增所有啟用的ExporterListener到ListenerExporterWrapper例項中，再經過ProtocolFilterWrapper處理，載入所有啟用的Filter，並且構建Invoker-Filter鏈包裝成新的Invoker。接著會呼叫DubboProtocol.export方法生成一個DubboExporter物件，Exporter中持有上面包裝Filter鏈後的Invoker物件引用和url對應的key（key的生成規則在com.alibaba.dubbo.rpc.support.ProtocolUtils.serviceKey(URL)方法中），該Exporter物件會註冊到DubboProtocol的exporterMap中，伺服器監聽到服務呼叫之後會在這個map中查詢Exporter，並封裝的Invoker。
com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
// export service.
String key = serviceKey(url);
DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
exporterMap.put(key, exporter);
//export an stub service for dispaching event
Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
if (isStubSupportEvent && !isCallbackservice){
String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
if (logger.isWarnEnabled()){
logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
"], has set stubproxy support event ,but no stub methods founded."));
}
} else {
stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
}
}
openServer(url);
return exporter;
}六、啟動服務監聽
上面已經生成了服務代理，為了能收到消費端的服務請求，還要在上面選中的埠上監聽消費端的請求，呼叫DubboProtocol的openServer方法，dubbo使用了Netty、grizzly、Mina三種網路框架，框架預設使用Netty，示例配置沒有特殊配置，主要看一下Netty。啟動服務時元件的呼叫順序：Exchanger.bind->Transporter.bind->Server，Exchanger和Transporter也是通過ExtensionLoader來載入，會載入到NettyTransporter實現，它會建立生成一個NettyServer例項，在構造NettyServer例項時呼叫doOpen開啟埠監聽， 並新增三個IO事件處理器，NettyCodecAdapter負責請求解碼、響應編碼，nettyHandler處理請求，把NettyCodecAdapter.decoder->NettyCodecAdapter.encoder->NettyHandler IO事件處理器鏈到Netty框架中，服務呼叫訊息由多層框架內部的Handler轉發最終會轉發到DubboProcotol的requestHandler中處理，服務的呼叫流程後面再詳細分析。
com.alibaba.dubbo.remoting.transport.netty.NettyServer.doOpen()
@Override
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
bootstrap = new ServerBootstrap(channelFactory);
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
channels = nettyHandler.getChannels();
// https://issues.jboss.org/browse/NETTY-365
// https://issues.jboss.org/browse/NETTY-379
// final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
/*int idleTimeout = getIdleTimeout();
if (idleTimeout > 10000) {
pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
}*/
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
// bind
channel = bootstrap.bind(getBindAddress());
}
七、服務註冊到註冊中心
服務起好之後，接下來是把提供者註冊到註冊中心，和註冊中心建立聯絡。回到RegistryProtocol.export方法，框架由RegistryFactory加載出一個Registry元件來處理註冊中心相關的邏輯。
/**
* 根據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);
}
registryFactory是啥時候設定進來的？dubbo框架在建立元件例項的時候，它會掃描所有的set方法，如果set方法的引數型別是可擴充套件的元件型別，會載入對應的擴充套件點例項並設定進去。
com.alibaba.dubbo.common.extension.ExtensionLoader
private T injectExtension(T instance) {
try {
if (objectFactory != null) {
for (Method method : instance.getClass().getMethods()) {
if (method.getName().startsWith("set")
&& method.getParameterTypes().length == 1
&& Modifier.isPublic(method.getModifiers())) {
Class<?> pt = method.getParameterTypes()[0];
try {
String property = method.getName().length() > 3 ? method.getName().substring(3, 4).toLowerCase() + method.getName().substring(4) : "";
Object object = objectFactory.getExtension(pt, property);
if (object != null) {
method.invoke(instance, object);
}
} catch (Exception e) {
logger.error("fail to inject via method " + method.getName()
+ " of interface " + type.getName() + ": " + e.getMessage(), e);
}
}
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return instance;
}Dubbo內建支援zookeeper、redis、廣播註冊中心。例項配置的zookeeper註冊中心，對應ZookeeperRegistry。提供者會在zookeeper註冊中心生成一個節點，節點路徑為/dubbo/interfaceClass/providers/interfaceClass/providers/{providerUrl}，儲存了服務提供方ip、埠、group、介面名稱、版本、應用名稱（redis註冊中心通過set命令把這些資訊快取到redis伺服器）。
com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry.doRegister(URL)
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);
}
}為了感知註冊中心的一些配置變化，提供者會監聽註冊中心路徑/dubbo/${interfaceClass}/configurators的節點，監聽該節點在註冊中心的一些配置資訊變更。zookeeper註冊中心通過zookeeper框架的監聽回撥介面進行監聽（redis註冊中心通過訂閱命令（subscribe）監聽），伺服器快取註冊中心的配置，當配置發生變更時，服務會重新整理本地快取，程式碼在ZookeeperRegistry的doSubscribe方法。
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>();
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);
}
}上面是服務提供者在初始化時做的一些主要動作，其實框架做的事遠比這多了，先暫時先忽略一些細節，專項功能放到後面再逐個擊破。
八、dubbo服務初始化與暴露服務總結
spring容器啟動時會解析dubbo的自定義標籤，解析的最終目的是返回 RootBeanDefinition
物件，RootBeanDefinition包含了解析出來的關於bean的所有資訊，注意，在bean的解析完後其實只是spring將其轉化成spring內部的一種抽象的資料物件結構，bean的建立（例項化）是第一次呼叫 getBean 時建立的。例如dubbo:service標籤被解析成一個ServiceBean，ServiceBean實現了ApplicationListener和InitializingBean介面，ServiceBean有個核心方法export，在這個方法中初始化服務提供者並且暴露遠端服務。這個方法在bean初始化或容器中所有bean重新整理完畢時被呼叫，根據provider的延遲設定決定，如果設定了延遲（delay屬性）在bean初始化結束之後呼叫否則容易重新整理事件中被呼叫，預設會延遲export，即在所有bean的重新整理結束時間中被呼叫。
生成url，
a.註冊中心的url：在服務初始化時需要對註冊中心進行訪問，所以需要一個URL來跟註冊中心通訊，根據應用的註冊中心配置來生成這個URL，URL協議是registry，包含了註冊中心地址、埠，path部分是com.alibaba.dubbo.registry.RegistryService類名，其中還有dubbo版本、應用名稱等資訊放在URL的引數中，格式化之後是這種形式：registry://10.165.124.205:2181/com.alibaba.dubbo.registry.RegistryService?application=…
b.服務本身的url:還需要生成一個URL來攜帶服務本身的資訊，協議由應用側指定，在我們的示例配置中是dubbo協議，host是本機ip，埠是上面選中的隨機或配置中指定的埠，path是服務對應的介面class全名（含包路徑），新增side引數provider，dubbo版本號、服務方法名稱、group等等，格式化之後是這種形式：
dubbo://10.240.176.159:20880/com.netease.haitao.mykaola.generic.api.InternalStaffServiceFacade?…&createInternalStaffOperDetail.timeout=10000&…&default.group=hzchenyunyun1&…&dubbo=3.0.0&…&methods=…&…&side=provider&… 生成遠端服務代理,生成代理的目的是你呼叫 invoker 的相關函式後，就會等於是呼叫 DubboInvoker 中的相關函式；也就是將本地呼叫轉為網路呼叫並獲得結果；
啟動服務監聽，框架預設使用Netty，開啟埠監聽，
並新增三個IO事件處理器，NettyCodecAdapter負責請求解碼、響應編碼，nettyHandler處理請求。暴露服務（將服務註冊到註冊中心），提供者會在zookeeper註冊中心生成一個節點，此節點儲存了服務提供方ip、埠、group、介面名稱、版本、應用名稱（redis註冊中心通過set命令把這些資訊快取到redis伺服器）。
為了感知註冊中心的一些配置變化，提供者會監聽註冊中心路徑/dubbo/${interfaceClass}/configurators的節點，監聽該節點在註冊中心的一些配置資訊變更。zookeeper註冊中心通過zookeeper框架的監聽回撥介面進行監聽（redis註冊中心通過訂閱命令（subscribe）監聽），伺服器快取註冊中心的配置，當配置發生變更時，服務會重新整理本地快取，程式碼在ZookeeperRegistry的doSubscribe方法。