1. 簡介
在
上一篇文章中，我詳細的分析了服務導出的原理。本篇文章我們趁熱打鐵，繼續分析服務引用的原理。在 Dubbo 中，我們可以通過兩種方式引用遠端服務。第一種是使用服務直聯的方式引用服務，第二種方式是基于注冊中心進行引用。服務直聯的方式僅适合在調試或測試服務的場景下使用，不适合線上上環境使用。是以，本文我将重點分析通過注冊中心引用服務的過程。從注冊中心中擷取服務配置隻是服務引用過程中的一環，除此之外，服務消費者還需要經曆 Invoker 建立、代理類建立等步驟。這些步驟，我将在後續章節中一一進行分析。
2.服務引用原理
Dubbo 服務引用的時機有兩個，第一個是在 Spring 容器調用 ReferenceBean 的 afterPropertiesSet 方法時引用服務，第二個是在 ReferenceBean 對應的服務被注入到其他類中時引用。這兩個引用服務的時機差別在于，第一個是餓漢式的，第二個是懶漢式的。預設情況下，Dubbo 使用懶漢式引用服務。如果需要使用餓漢式，可通過配置 <dubbo:reference> 的 init 屬性開啟。下面我們按照 Dubbo 預設配置進行分析，整個分析過程從 ReferenceBean 的 getObject 方法開始。當我們的服務被注入到其他類中時，Spring 會第一時間調用 getObject 方法，并由該方法執行服務引用邏輯。按照慣例，在進行具體工作之前，需先進行配置檢查與收集工作。接着根據收集到的資訊決定服務用的方式，有三種，第一種是引用本地 (JVM) 服務，第二是通過直聯方式引用遠端服務，第三是通過注冊中心引用遠端服務。不管是哪種引用方式，最後都會得到一個 Invoker 執行個體。如果有多個注冊中心，多個服務提供者，這個時候會得到一組 Invoker 執行個體，此時需要通過叢集管理類 Cluster 将多個 Invoker 合并成一個執行個體。合并後的 Invoker 執行個體已經具備調用本地或遠端服務的能力了，但并不能将此執行個體暴露給使用者使用，這會對使用者業務代碼造成侵入。此時架構還需要通過代理工廠類 (ProxyFactory) 為服務接口生成代理類，并讓代理類去調用 Invoker 邏輯。避免了 Dubbo 架構代碼對業務代碼的侵入，同時也讓架構更容易使用。
以上就是 Dubbo 引用服務的大緻原理，下面我們深入到代碼中，詳細分析服務引用細節。
3.源碼分析
服務引用的入口方法為 ReferenceBean 的 getObject 方法，該方法定義在 Spring 的 FactoryBean 接口中，ReferenceBean 實作了這個方法。實作代碼如下：
public Object getObject() throws Exception {
return get();
}
public synchronized T get() {
if (destroyed) {
throw new IllegalStateException("Already destroyed!");
}
// 檢測 ref 是否為空，為空則通過 init 方法建立
if (ref == null) {
// init 方法主要用于處理配置，以及調用 createProxy 生成代理類
init();
}
return ref;
} 這裡兩個方法代碼都比較簡短，并不難了解。不過這裡需要特别說明一下，如果大家從 getObject 方法進行代碼調試時，會碰到比較詫異的問題。這裡假設你使用 IDEA，且保持了 IDEA 的預設配置。當你面調試到 get 方法的
if (ref == null)
時，你會驚奇的發現 ref 不為空，導緻你無法進入到 init 方法中繼續調試。導緻這個現象的原因是 Dubbo 架構本身有點小問題，這個小問題會引發一些讓人詫異的現象。關于這個問題，我進行了将近兩個小時的排查。查明問題後，我給 Dubbo 送出了一個 pull request (
#2754) 介紹這個問題，有興趣的朋友可以去看看。大家如果想規避這個問題，可以修改一下 IDEA 的配置。在配置面闆中搜尋 toString，然後取消
Enable 'toString' object view
前的對号。具體如下：
講完需要注意的點，我們繼續向下分析，接下來将分析配置的處理過程。
3.1 處理配置
Dubbo 提供了豐富的配置，用于調整和優化架構行為，性能等。Dubbo 在引用或導出服務時，首先會對這些配置進行檢查和處理，以保證配置到正确性。如果大家不是很熟悉 Dubbo 配置，建議先閱讀以下官方文檔。配置解析的方法為 ReferenceConfig 的 init 方法，下面來看一下方法邏輯。
private void init() {
if (initialized) {
return;
}
initialized = true;
if (interfaceName == null || interfaceName.length() == 0) {
throw new IllegalStateException("interface not allow null!");
}
// 檢測 consumer 變量是否為空，為空則建立
checkDefault();
appendProperties(this);
if (getGeneric() == null && getConsumer() != null) {
// 設定 generic
setGeneric(getConsumer().getGeneric());
}
// 檢測是否為泛化接口
if (ProtocolUtils.isGeneric(getGeneric())) {
interfaceClass = GenericService.class;
} else {
try {
// 加載類
interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
.getContextClassLoader());
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
checkInterfaceAndMethods(interfaceClass, methods);
}
// ------------------------------- 分割線1 ------------------------------
// 從系統變量中擷取與接口名對應的屬性值
String resolve = System.getProperty(interfaceName);
String resolveFile = null;
if (resolve == null || resolve.length() == 0) {
// 從系統屬性中擷取解析檔案路徑
resolveFile = System.getProperty("dubbo.resolve.file");
if (resolveFile == null || resolveFile.length() == 0) {
// 從指定位置加載配置檔案
File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");
if (userResolveFile.exists()) {
// 擷取檔案絕對路徑
resolveFile = userResolveFile.getAbsolutePath();
}
}
if (resolveFile != null && resolveFile.length() > 0) {
Properties properties = new Properties();
FileInputStream fis = null;
try {
fis = new FileInputStream(new File(resolveFile));
// 從檔案中加載配置
properties.load(fis);
} catch (IOException e) {
throw new IllegalStateException("Unload ..., cause:...");
} finally {
try {
if (null != fis) fis.close();
} catch (IOException e) {
logger.warn(e.getMessage(), e);
}
}
// 擷取與接口名對應的配置
resolve = properties.getProperty(interfaceName);
}
}
if (resolve != null && resolve.length() > 0) {
// 将 resolve 指派給 url
url = resolve;
}
// ------------------------------- 分割線2 ------------------------------
if (consumer != null) {
if (application == null) {
// 從 consumer 中擷取 Application 執行個體，下同
application = consumer.getApplication();
}
if (module == null) {
module = consumer.getModule();
}
if (registries == null) {
registries = consumer.getRegistries();
}
if (monitor == null) {
monitor = consumer.getMonitor();
}
}
if (module != null) {
if (registries == null) {
registries = module.getRegistries();
}
if (monitor == null) {
monitor = module.getMonitor();
}
}
if (application != null) {
if (registries == null) {
registries = application.getRegistries();
}
if (monitor == null) {
monitor = application.getMonitor();
}
}
// 檢測本地 Application 和本地存根配置合法性
checkApplication();
checkStubAndMock(interfaceClass);
// ------------------------------- 分割線3 ------------------------------
Map<String, String> map = new HashMap<String, String>();
Map<Object, Object> attributes = new HashMap<Object, Object>();
// 添加 side、協定版本資訊、時間戳和程序号等資訊到 map 中
map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
map.put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());
map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
if (ConfigUtils.getPid() > 0) {
map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
}
if (!isGeneric()) { // 非泛化服務
// 擷取版本
String revision = Version.getVersion(interfaceClass, version);
if (revision != null && revision.length() > 0) {
map.put("revision", revision);
}
// 擷取接口方法清單，并添加到 map 中
String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
if (methods.length == 0) {
map.put("methods", Constants.ANY_VALUE);
} else {
map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
}
}
map.put(Constants.INTERFACE_KEY, interfaceName);
// 将 ApplicationConfig、ConsumerConfig、ReferenceConfig 等對象的字段資訊添加到 map 中
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, this);
// ------------------------------- 分割線4 ------------------------------
String prefix = StringUtils.getServiceKey(map);
if (methods != null && !methods.isEmpty()) {
// 周遊 MethodConfig 清單
for (MethodConfig method : methods) {
appendParameters(map, method, method.getName());
String retryKey = method.getName() + ".retry";
// 檢測 map 是否包含 methodName.retry
if (map.containsKey(retryKey)) {
String retryValue = map.remove(retryKey);
if ("false".equals(retryValue)) {
// 添加重試次數配置 methodName.retries
map.put(method.getName() + ".retries", "0");
}
}
// 添加 MethodConfig 中的“屬性”字段到 attributes
// 比如 onreturn、onthrow、oninvoke 等
appendAttributes(attributes, method, prefix + "." + method.getName());
checkAndConvertImplicitConfig(method, map, attributes);
}
}
// ------------------------------- 分割線5 ------------------------------
// 擷取服務消費者 ip 位址
String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
if (hostToRegistry == null || hostToRegistry.length() == 0) {
hostToRegistry = NetUtils.getLocalHost();
} else if (isInvalidLocalHost(hostToRegistry)) {
throw new IllegalArgumentException("Specified invalid registry ip from property..." );
}
map.put(Constants.REGISTER_IP_KEY, hostToRegistry);
// 存儲 attributes 到系統上下文中
StaticContext.getSystemContext().putAll(attributes);
// 建立代理類
ref = createProxy(map);
// 根據服務名，ReferenceConfig，代理類建構 ConsumerModel，
// 并将 ConsumerModel 存入到 ApplicationModel 中
ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
} 上面的代碼很長，做的事情比較多。這裡我根據代碼邏輯，對代碼進行了分塊，下面我們一起來看一下。
首先是方法開始到分割線1之間的代碼。這段代碼主要用于檢測 ConsumerConfig 執行個體是否存在，如不存在則建立一個新的執行個體，然後通過系統變量或 dubbo.properties 配置檔案填充 ConsumerConfig 的字段。接着是檢測泛化配置，并根據配置設定 interfaceClass 的值。本段代碼邏輯大緻就是這些，接着來看分割線1到分割線2之間的邏輯。這段邏輯用于從系統屬性或配置檔案中加載與接口名相對應的配置，并将解析結果指派給 url 字段。url 字段的作用一般是用于點對點調用。繼續向下看，分割線2和分割線3之間的代碼用于檢測幾個核心配置類是否為空，為空則嘗試從其他配置類中擷取。分割線3與分割線4之間的代碼主要是用于收集各種配置，并将配置存儲到 map 中。分割線4和分割線5之間的代碼用于處理 MethodConfig 執行個體。該執行個體包含了事件通知配置，比如 onreturn、onthrow、oninvoke 等。分割線5到方法結尾的代碼主要用于解析服務消費者 ip，以及調用 createProxy 建立代理對象。關于該方法的詳細分析，将會在接下來的章節中展開。
到這裡，關于配置的檢查與處理過長就分析完了。這部分邏輯不是很難了解，但比較繁雜，大家需要耐心看一下。好了，本節先到這，接下來分析服務引用的過程。
3.2 引用服務
本節我們要從 createProxy 開始看起。createProxy 這個方法表面上看起來隻是用于建立代理對象，但實際上并非如此。該方法還會調用其他方法建構以及合并 Invoker 執行個體。具體細節如下。
private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp", "localhost", 0, map);
final boolean isJvmRefer;
if (isInjvm() == null) {
// url 配置被指定，則不做本地引用
if (url != null && url.length() > 0) {
isJvmRefer = false;
// 根據 url 的協定、scope 以及 injvm 等參數檢測是否需要本地引用
// 比如如果使用者顯式配置了 scope=local，此時 isInjvmRefer 傳回 true
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
isJvmRefer = true;
} else {
isJvmRefer = false;
}
} else {
// 擷取 injvm 配置值
isJvmRefer = isInjvm().booleanValue();
}
// 本地引用
if (isJvmRefer) {
// 生成本地引用 URL，協定為 injvm
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
// 調用 refer 方法建構 InjvmInvoker 執行個體
invoker = refprotocol.refer(interfaceClass, url);
// 遠端引用
} else {
// url 不為空，表明使用者可能想進行點對點調用
if (url != null && url.length() > 0) {
// 當需要配置多個 url 時，可用分号進行分割，這裡會進行切分
String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (url.getPath() == null || url.getPath().length() == 0) {
// 設定接口全限定名為 url 路徑
url = url.setPath(interfaceName);
}
// 檢測 url 協定是否為 registry，若是，表明使用者想使用指定的注冊中心
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
// 将 map 轉換為查詢字元串，并作為 refer 參數的值添加到 url 中
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else {
// 合并 url，移除服務提供者的一些配置（這些配置來源于使用者配置的 url 屬性），
// 比如線程池相關配置。并保留服務提供者的部配置設定置，比如版本，group，時間戳等
// 最後将合并後的配置設定為 url 查詢字元串中。
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else {
// 加載注冊中心 url
List<URL> us = loadRegistries(false);
if (us != null && !us.isEmpty()) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if (monitorUrl != null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
// 添加 refer 參數到 url 中，并将 url 添加到 urls 中
urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
}
}
// 未配置注冊中心，抛出異常
if (urls.isEmpty()) {
throw new IllegalStateException("No such any registry to reference...");
}
}
// 單個注冊中心或服務提供者(服務直聯，下同)
if (urls.size() == 1) {
// 調用 RegistryProtocol 的 refer 建構 Invoker 執行個體
invoker = refprotocol.refer(interfaceClass, urls.get(0));
// 多個注冊中心或多個服務提供者，或者兩者混合
} else {
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
// 擷取所有的 Invoker
for (URL url : urls) {
// 通過 refprotocol 調用 refer 建構 Invoker，refprotocol 會在運作時
// 根據 url 協定頭加載指定的 Protocol 執行個體，并調用執行個體的 refer 方法
invokers.add(refprotocol.refer(interfaceClass, url));
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url;
}
}
if (registryURL != null) {
// 如果注冊中心連結不為空，則将使用 AvailableCluster
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
// 建立 StaticDirectory 執行個體，并由 Cluster 對多個 Invoker 進行合并
invoker = cluster.join(new StaticDirectory(u, invokers));
} else {
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
Boolean c = check;
if (c == null && consumer != null) {
c = consumer.isCheck();
}
if (c == null) {
c = true;
}
// invoker 可用性檢查
if (c && !invoker.isAvailable()) {
throw new IllegalStateException("No provider available for the service...");
}
// 生成代理類
return (T) proxyFactory.getProxy(invoker);
} 上面代碼很多，不過邏輯比較清晰。首先根據配置檢查是否為本地調用，若是，則調用 InjvmProtocol 的 refer 方法生成 InjvmInvoker 執行個體。若不是，則讀取直聯配置項，或注冊中心 url，并将讀取到的 url 存儲到 urls 中。然後，根據 urls 元素數量進行後續操作。若 urls 元素數量為1，則直接通過 Protocol 自适應拓展建構 Invoker 執行個體接口。若 urls 元素數量大于1，即存在多個注冊中心或服務直聯 url，此時先根據 url 建構 Invoker。然後再通過 Cluster 合并多個 Invoker，最後調用 ProxyFactory 生成代理類。這裡，Invoker 的建構過程以及代理類的過程比較重要，是以我将分兩小節對這兩個過程進行分析。
3.2.1 建立 Invoker
Invoker 是 Dubbo 的核心模型，代表一個可執行體。在服務提供方，Invoker 用于調用服務提供類。在服務消費方，Invoker 用于執行遠端調用。Invoker 在 Dubbo 中的位置十分重要，是以我們有必要去搞懂它。從前面的代碼中可知，Invoker 是由 Protocol 實作類建構的。Protocol 實作類有很多，這裡我會分析最常用的兩個，分别是 RegistryProtocol 和 DubboProtocol，其他的大家自行分析。下面先來分析 DubboProtocol 的 refer 方法源碼。如下：
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
optimizeSerialization(url);
// 建立 DubboInvoker
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
} 上面方法看起來比較簡單，不過這裡有一個調用需要我們注意一下，即 getClients。這個方法用于擷取用戶端執行個體，執行個體類型為 ExchangeClient。ExchangeClient 實際上并不具備通信能力，是以它需要更底層的用戶端執行個體進行通信。比如 NettyClient、MinaClient 等，預設情況下，Dubbo 使用 NettyClient 進行通信。接下來，我們簡單看一下 getClients 方法的邏輯。
private ExchangeClient[] getClients(URL url) {
// 是否共享連接配接
boolean service_share_connect = false;
// 擷取連接配接數，預設為0，表示未配置
int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
// 如果未配置 connections，則共享連接配接
if (connections == 0) {
service_share_connect = true;
connections = 1;
}
ExchangeClient[] clients = new ExchangeClient[connections];
for (int i = 0; i < clients.length; i++) {
if (service_share_connect) {
// 擷取共享用戶端
clients[i] = getSharedClient(url);
} else {
// 初始化新的用戶端
clients[i] = initClient(url);
}
}
return clients;
} 這裡根據 connections 數量決定是擷取共享用戶端還是建立新的用戶端執行個體，預設情況下，使用共享用戶端執行個體。不過 getSharedClient 方法中也會調用 initClient 方法，是以下面我們一起看一下這兩個方法。
private ExchangeClient getSharedClient(URL url) {
String key = url.getAddress();
// 擷取帶有“引用計數”功能的 ExchangeClient
ReferenceCountExchangeClient client = referenceClientMap.get(key);
if (client != null) {
if (!client.isClosed()) {
// 增加引用計數
client.incrementAndGetCount();
return client;
} else {
referenceClientMap.remove(key);
}
}
locks.putIfAbsent(key, new Object());
synchronized (locks.get(key)) {
if (referenceClientMap.containsKey(key)) {
return referenceClientMap.get(key);
}
// 建立 ExchangeClient 用戶端
ExchangeClient exchangeClient = initClient(url);
// 将 ExchangeClient 執行個體傳給 ReferenceCountExchangeClient，這裡明顯用了裝飾模式
client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
referenceClientMap.put(key, client);
ghostClientMap.remove(key);
locks.remove(key);
return client;
}
} 上面方法先通路緩存，若緩存未命中，則通過 initClient 方法建立新的 ExchangeClient 執行個體，并将該執行個體傳給 ReferenceCountExchangeClient 構造方法建立一個帶有引用技術功能的 ExchangeClient 執行個體。ReferenceCountExchangeClient 内部實作比較簡單，就不分析了。下面我們再來看一下 initClient 方法的代碼。
private ExchangeClient initClient(URL url) {
// 擷取用戶端類型，預設為 netty
String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
// 添加編解碼和心跳包參數到 url 中
url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
// 檢測用戶端類型是否存在，不存在則抛出異常
if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
throw new RpcException("Unsupported client type: ...");
}
ExchangeClient client;
try {
// 擷取 lazy 配置，并根據配置值決定建立的用戶端類型
if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
// 建立懶加載 ExchangeClient 執行個體
client = new LazyConnectExchangeClient(url, requestHandler);
} else {
// 建立普通 ExchangeClient 執行個體
client = Exchangers.connect(url, requestHandler);
}
} catch (RemotingException e) {
throw new RpcException("Fail to create remoting client for service...");
}
return client;
} initClient 方法首先擷取使用者配置的用戶端類型，預設為 netty。然後檢測使用者配置的用戶端類型是否存在，不存在則抛出異常。最後根據 lazy 配置決定建立什麼類型的用戶端。這裡的 LazyConnectExchangeClient 代碼并不是很複雜，該類會在 request 方法被調用時通過 Exchangers 的 connect 方法建立 ExchangeClient 用戶端，這裡就不分析 LazyConnectExchangeClient 的代碼了。下面我們分析一下 Exchangers 的 connect 方法。
public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
// 擷取 Exchanger 執行個體，預設為 HeaderExchangeClient
return getExchanger(url).connect(url, handler);
} 如上，getExchanger 會通過 SPI 加載 HeaderExchangeClient 執行個體，這個方法比較簡單，大家自己看一下吧。接下來分析 HeaderExchangeClient 的實作。
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
// 這裡包含了多個調用，分别如下：
// 1. 建立 HeaderExchangeHandler 對象
// 2. 建立 DecodeHandler 對象
// 3. 通過 Transporters 建構 Client 執行個體
// 4. 建立 HeaderExchangeClient 對象
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
} 這裡的調用比較多，我們這裡重點看一下 Transporters 的 connect 方法。如下：
public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
ChannelHandler handler;
if (handlers == null || handlers.length == 0) {
handler = new ChannelHandlerAdapter();
} else if (handlers.length == 1) {
handler = handlers[0];
} else {
// 如果 handler 數量大于1，則建立一個 ChannelHandler 分發器
handler = new ChannelHandlerDispatcher(handlers);
}
// 擷取 Transporter 自适應拓展類，并調用 connect 方法生成 Client 執行個體
return getTransporter().connect(url, handler);
} 這裡，getTransporter 方法傳回的是自适應拓展類，該類會在運作時根據用戶端類型加載指定的 Transporter 實作類。若使用者未顯示配置用戶端類型，則預設加載 NettyTransporter，并調用該類的 connect 方法。如下：
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
// 建立 NettyClient 對象
return new NettyClient(url, listener);
} 到這裡就不繼續跟下去了，在往下就是通過 Netty 提供的接口建構 Netty 用戶端了，大家有興趣自己看看。到這裡，關于 DubboProtocol 的 refer 方法就分析完了。接下來，繼續分析 RegistryProtocol 的 refer 方法邏輯。
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 取 registry 參數值，并将其設定為協定頭
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
// 擷取注冊中心執行個體
Registry registry = registryFactory.getRegistry(url);
// 這個判斷暫時不知道有什麼意圖，為什麼要給 RegistryService 類型生成 Invoker ？
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// 将 url 查詢字元串轉為 Map
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
// 擷取 group 配置
String group = qs.get(Constants.GROUP_KEY);
if (group != null && group.length() > 0) {
if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
|| "*".equals(group)) {
// 通過 SPI 加載 MergeableCluster 執行個體，并調用 doRefer 繼續執行引用服務邏輯
return doRefer(getMergeableCluster(), registry, type, url);
}
}
// 調用 doRefer 繼續執行引用服務邏輯
return doRefer(cluster, registry, type, url);
} 上面代碼首先為 url 設定協定頭，然後根據 url 參數加載注冊中心執行個體。接下來對 RegistryService 繼續針對性處理，這個處理邏輯我不是很明白，不知道為什麼要為 RegistryService 類型生成 Invoker，有知道同學麻煩告知一下。然後就是擷取 group 配置，根據 group 配置決定 doRefer 第一個參數的類型。這裡的重點是 doRefer 方法，如下：
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// 建立 RegistryDirectory 執行個體
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
// 設定注冊中心和協定
directory.setRegistry(registry);
directory.setProtocol(protocol);
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);
// 注冊服務消費者，在 consumers 目錄下新節點
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)));
}
// 訂閱 providers、configurators、routers 等節點資料
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
// 一個注冊中心可能有多個服務提供者，是以這裡需要将多個服務提供者合并為一個
Invoker invoker = cluster.join(directory);
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
return invoker;
} 如上，doRefer 方法建立一個 RegistryDirectory 執行個體，然後生成服務者消費者連結，并向注冊中心進行注冊。注冊完畢後，緊接着訂閱 providers、configurators、routers 等節點下的資料。完成訂閱後，RegistryDirectory 會收到這幾個節點下的子節點資訊，比如可以擷取到服務提供者的配置資訊。由于一個服務可能部署在多台伺服器上，這樣就會在 providers 産生多個節點，這個時候就需要 Cluster 将多個服務節點合并為一個，并生成一個 Invoker。關于 RegistryDirectory 和 Cluster，本文不打算進行分析，相關分析将會在随後的文章中展開。
好了，關于 Invoker 的建立的邏輯就先分析到這。邏輯比較多，大家耐心看一下。
3.2.2 建立代理
Invoker 建立完畢後，接下來要做的事情是為服務接口生成代理對象。有了代理對象，我們就可以通過代理對象進行遠端調用。代理對象生成的入口方法為在 ProxyFactory 的 getProxy，接下來進行分析。
public <T> T getProxy(Invoker<T> invoker) throws RpcException {
// 調用重載方法
return getProxy(invoker, false);
}
public <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException {
Class<?>[] interfaces = null;
// 擷取接口清單
String config = invoker.getUrl().getParameter("interfaces");
if (config != null && config.length() > 0) {
// 切分接口清單
String[] types = Constants.COMMA_SPLIT_PATTERN.split(config);
if (types != null && types.length > 0) {
interfaces = new Class<?>[types.length + 2];
// 設定服務接口類和 EchoService.class 到 interfaces 中
interfaces[0] = invoker.getInterface();
interfaces[1] = EchoService.class;
for (int i = 0; i < types.length; i++) {
// 加載接口類
interfaces[i + 1] = ReflectUtils.forName(types[i]);
}
}
}
if (interfaces == null) {
interfaces = new Class<?>[]{invoker.getInterface(), EchoService.class};
}
// 為 http 和 hessian 協定提供泛化調用支援，參考 pull request #1827
if (!invoker.getInterface().equals(GenericService.class) && generic) {
int len = interfaces.length;
Class<?>[] temp = interfaces;
// 建立新的 interfaces 數組
interfaces = new Class<?>[len + 1];
System.arraycopy(temp, 0, interfaces, 0, len);
// 設定 GenericService.class 到數組中
interfaces[len] = GenericService.class;
}
// 調用重載方法
return getProxy(invoker, interfaces);
}
public abstract <T> T getProxy(Invoker<T> invoker, Class<?>[] types); 如上，上面大段代碼都是用來擷取 interfaces 數組的，是以我們需要繼續往下看。getProxy(Invoker, Class<?>[]) 這個方法是一個抽象方法，下面我們到 JavassistProxyFactory 類中看一下該方法的實作代碼。
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
// 生成 Proxy 子類（Proxy 是抽象類）。并調用Proxy 子類的 newInstance 方法生成 Proxy 執行個體
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
} 上面代碼并不多，首先是通過 Proxy 的 getProxy 方法擷取 Proxy 子類，然後建立 InvokerInvocationHandler 對象，并将該對象傳給 newInstance 生成 Proxy 執行個體。InvokerInvocationHandler 實作自 JDK 的 InvocationHandler 接口，具體的用途是攔截接口類調用。該類邏輯比較簡單，這裡就不分析了。下面我們重點關注一下 Proxy 的 getProxy 方法，如下。
public static Proxy getProxy(Class<?>... ics) {
// 調用重載方法
return getProxy(ClassHelper.getClassLoader(Proxy.class), ics);
}
public static Proxy getProxy(ClassLoader cl, Class<?>... ics) {
if (ics.length > 65535)
throw new IllegalArgumentException("interface limit exceeded");
StringBuilder sb = new StringBuilder();
// 周遊接口清單
for (int i = 0; i < ics.length; i++) {
String itf = ics[i].getName();
// 檢測類型是否為接口
if (!ics[i].isInterface())
throw new RuntimeException(itf + " is not a interface.");
Class<?> tmp = null;
try {
// 重新加載接口類
tmp = Class.forName(itf, false, cl);
} catch (ClassNotFoundException e) {
}
// 檢測接口是否相同，這裡 tmp 有可能為空
if (tmp != ics[i])
throw new IllegalArgumentException(ics[i] + " is not visible from class loader");
// 拼接接口全限定名，分隔符為 ;
sb.append(itf).append(';');
}
// 使用拼接後接口名作為 key
String key = sb.toString();
Map<String, Object> cache;
synchronized (ProxyCacheMap) {
cache = ProxyCacheMap.get(cl);
if (cache == null) {
cache = new HashMap<String, Object>();
ProxyCacheMap.put(cl, cache);
}
}
Proxy proxy = null;
synchronized (cache) {
do {
// 從緩存中擷取 Reference<Proxy> 執行個體
Object value = cache.get(key);
if (value instanceof Reference<?>) {
proxy = (Proxy) ((Reference<?>) value).get();
if (proxy != null) {
return proxy;
}
}
// 多線程控制，保證隻有一個線程可以進行後續操作
if (value == PendingGenerationMarker) {
try {
// 其他線程在此處進行等待
cache.wait();
} catch (InterruptedException e) {
}
} else {
// 放置标志位到緩存中，并跳出 while 循環進行後續操作
cache.put(key, PendingGenerationMarker);
break;
}
}
while (true);
}
long id = PROXY_CLASS_COUNTER.getAndIncrement();
String pkg = null;
ClassGenerator ccp = null, ccm = null;
try {
// 建立 ClassGenerator 對象
ccp = ClassGenerator.newInstance(cl);
Set<String> worked = new HashSet<String>();
List<Method> methods = new ArrayList<Method>();
for (int i = 0; i < ics.length; i++) {
// 檢測接口通路級别是否為 protected 或 privete
if (!Modifier.isPublic(ics[i].getModifiers())) {
// 擷取接口包名
String npkg = ics[i].getPackage().getName();
if (pkg == null) {
pkg = npkg;
} else {
if (!pkg.equals(npkg))
// 非 public 級别的接口必須在同一個包下，否者抛出異常
throw new IllegalArgumentException("non-public interfaces from different packages");
}
}
// 添加接口到 ClassGenerator 中
ccp.addInterface(ics[i]);
// 周遊接口方法
for (Method method : ics[i].getMethods()) {
// 擷取方法描述，可了解為方法簽名
String desc = ReflectUtils.getDesc(method);
// 如果已包含在 worked 中，則忽略。考慮這種情況，
// A 接口和 B 接口中包含一個完全相同的方法
if (worked.contains(desc))
continue;
worked.add(desc);
int ix = methods.size();
// 擷取方法傳回值類型
Class<?> rt = method.getReturnType();
// 擷取參數清單
Class<?>[] pts = method.getParameterTypes();
// 生成 Object[] args = new Object[1...N]
StringBuilder code = new StringBuilder("Object[] args = new Object[").append(pts.length).append("];");
for (int j = 0; j < pts.length; j++)
// 生成 args[1...N] = ($w)$1...N;
code.append(" args[").append(j).append("] = ($w)$").append(j + 1).append(";");
// 生成 InvokerHandler 接口的 invoker 方法調用語句，如下：
// Object ret = handler.invoke(this, methods[1...N], args);
code.append(" Object ret = handler.invoke(this, methods[" + ix + "], args);");
// 傳回值不為 void
if (!Void.TYPE.equals(rt))
// 生成傳回語句，形如 return (java.lang.String) ret;
code.append(" return ").append(asArgument(rt, "ret")).append(";");
methods.add(method);
// 添加方法名、通路控制符、參數清單、方法代碼等資訊到 ClassGenerator 中
ccp.addMethod(method.getName(), method.getModifiers(), rt, pts, method.getExceptionTypes(), code.toString());
}
}
if (pkg == null)
pkg = PACKAGE_NAME;
// 建構接口代理類名稱：pkg + ".proxy" + id，比如 com.tianxiaobo.proxy0
String pcn = pkg + ".proxy" + id;
ccp.setClassName(pcn);
ccp.addField("public static java.lang.reflect.Method[] methods;");
// 生成 private java.lang.reflect.InvocationHandler handler;
ccp.addField("private " + InvocationHandler.class.getName() + " handler;");
// 為接口代理類添加帶有 InvocationHandler 參數的構造方法，比如：
// porxy0(java.lang.reflect.InvocationHandler arg0) {
// handler=$1;
// }
ccp.addConstructor(Modifier.PUBLIC, new Class<?>[]{InvocationHandler.class}, new Class<?>[0], "handler=$1;");
// 為接口代理類添加預設構造方法
ccp.addDefaultConstructor();
// 生成接口代理類
Class<?> clazz = ccp.toClass();
clazz.getField("methods").set(null, methods.toArray(new Method[0]));
// 建構 Proxy 子類名稱，比如 Proxy1，Proxy2 等
String fcn = Proxy.class.getName() + id;
ccm = ClassGenerator.newInstance(cl);
ccm.setClassName(fcn);
ccm.addDefaultConstructor();
ccm.setSuperClass(Proxy.class);
// 為 Proxy 的抽象方法 newInstance 生成實作代碼，形如：
// public Object newInstance(java.lang.reflect.InvocationHandler h) {
// return new com.tianxiaobo.proxy0($1);
// }
ccm.addMethod("public Object newInstance(" + InvocationHandler.class.getName() + " h){ return new " + pcn + "($1); }");
// 生成 Proxy 實作類
Class<?> pc = ccm.toClass();
// 通過反射建立 Proxy 執行個體
proxy = (Proxy) pc.newInstance();
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
} finally {
if (ccp != null)
// 釋放資源
ccp.release();
if (ccm != null)
ccm.release();
synchronized (cache) {
if (proxy == null)
cache.remove(key);
else
// 寫緩存
cache.put(key, new WeakReference<Proxy>(proxy));
// 喚醒其他等待線程
cache.notifyAll();
}
}
return proxy;
} 上面代碼比較複雜，我也寫了很多注釋。大家在閱讀這段代碼時，要搞清楚 ccp 和 ccm 的用途，不然會被搞暈。ccp 用于為服務接口生成代理類，比如我們有一個 DemoService 接口，這個接口代理類就是由 ccp 生成的。ccm 則是用于為 org.apache.dubbo.common.bytecode.Proxy 抽象類生成子類，主要是實作 Proxy 的抽象方法。下面以 org.apache.dubbo.demo.DemoService 這個接口為例，來看一下該接口代理類代碼大緻是怎樣的（忽略 EchoService 接口）。
package org.apache.dubbo.common.bytecode;
public class proxy0 implements org.apache.dubbo.demo.DemoService {
public static java.lang.reflect.Method[] methods;
private java.lang.reflect.InvocationHandler handler;
public proxy0() {
}
public proxy0(java.lang.reflect.InvocationHandler arg0) {
handler = $1;
}
public java.lang.String sayHello(java.lang.String arg0) {
Object[] args = new Object[1];
args[0] = ($w) $1;
Object ret = handler.invoke(this, methods[0], args);
return (java.lang.String) ret;
}
} 好了，到這裡代理類生成邏輯就分析完了。整個過程比較複雜，大家需要耐心看一下，本節點到這裡。
4.總結
本篇文章對服務引用的過程進行了較為詳盡的分析，之是以說是較為詳盡，是因為還有一些地方沒有分析到。比如 Directory、Cluster 等實作類的代碼并未進行詳細分析，由于這些類功能比較獨立，是以我打算後續單獨成文進行分析。暫時我們可以先把這些類看成黑盒，隻要知道這些類的用途即可。引用服務過程涉及到的調用也非常多，大家在閱讀相關代碼的中耐心些，并多進行調試。
好了，本篇文章就先到這裡了。謝謝閱讀。
本文在知識共享許可協定 4.0 下釋出，轉載需在明顯位置處注明出處
作者：田小波
本文同步釋出在我的個人部落格：
http://www.tianxiaobo.com
本作品采用
知識共享署名-非商業性使用-禁止演繹 4.0 國際許可協定進行許可。