此文轉(zhuǎn)自 原文 ，在最后做了一點補充
目錄
1.簡介
2.服務(wù)引用原理
3.源碼分析
3.1 處置配置
3.2 引用服務(wù)
3.2.1 創(chuàng)建 Invoker
3.2.2 創(chuàng)建代理
4.總結(jié)
補充
服務(wù)引用調(diào)用鏈
1.簡介
在 Dubbo 中，我們可以通過兩種方式引用遠程服務(wù)。第一種是使用服務(wù)直連的方式引用服務(wù)，第二種方式是基于注冊中心進行引用。服務(wù)直連的方式僅適合在調(diào)試或測試服務(wù)的場景下使用，不適合在線上環(huán)境使用。因此，本文我將重點分析通過注冊中心引用服務(wù)的過程。從注冊中心中獲取服務(wù)配置只是服務(wù)引用過程中的一環(huán)，除此之外，服務(wù)消費者還需要經(jīng)歷 Invoker 創(chuàng)建、代理類創(chuàng)建等步驟。
2.服務(wù)引用原理
Dubbo 服務(wù)引用的時機有兩個：
第一個是在 Spring 容器調(diào)用 ReferenceBean 的 afterPropertiesSet 方法時引用服務(wù)，
第二個是在 ReferenceBean 對應(yīng)的服務(wù)被注入到其他類中時引用。
這兩個引用服務(wù)的時機區(qū)別在于，第一個是餓漢式的，第二個是懶漢式的。默認情況下，Dubbo 使用懶漢式引用服務(wù)。如果需要使用餓漢式，可通過配置 <dubbo:reference> 的 init 屬性開啟。下面我們按照 Dubbo 默認配置進行分析，整個分析過程從 ReferenceBean 的 getObject 方法開始。當我們的服務(wù)被注入到其他類中時，Spring 會第一時間調(diào)用 getObject 方法，并由該方法執(zhí)行服務(wù)引用邏輯。按照慣例，在進行具體工作之前，需先進行配置檢查與收集工作。
接著根據(jù)收集到的信息決定服務(wù)y引用的方式:
第一種是引用本地 (JVM) 服務(wù)，
第二是通過直連方式引用遠程服務(wù)，
第三是通過注冊中心引用遠程服務(wù)。
不管是哪種引用方式，最后都會得到一個 Invoker 實例。如果有多個注冊中心，多個服務(wù)提供者，這個時候會得到一組 Invoker 實例，此時需要通過集群管理類 Cluster 將多個 Invoker 合并成一個實例。合并后的 Invoker 實例已經(jīng)具備調(diào)用本地或遠程服務(wù)的能力了，但并不能將此實例暴露給用戶使用，這會對用戶業(yè)務(wù)代碼造成侵入。此時框架還需要通過代理工廠類 (ProxyFactory) 為服務(wù)接口生成代理類，并讓代理類去調(diào)用 Invoker 邏輯。避免了 Dubbo 框架代碼對業(yè)務(wù)代碼的侵入，同時也讓框架更容易使用。
3.源碼分析
服務(wù)引用的入口方法為 ReferenceBean 的 getObject 方法，該方法定義在 Spring 的 FactoryBean 接口中，ReferenceBean 實現(xiàn)了這個方法。實現(xiàn)代碼如下：
public Object getObject() throws Exception {
return get();
}
public synchronized T get() {
if (destroyed) {
throw new IllegalStateException("Already destroyed!");
}
// 檢測 ref 是否為空，為空則通過 init 方法創(chuàng)建
if (ref == null) {
// init 方法主要用于處理配置，以及調(diào)用 createProxy 生成代理類
init();
}
return ref;
}
3.1 處置配置
Dubbo 提供了豐富的配置，用于調(diào)整和優(yōu)化框架行為，性能等。Dubbo 在引用或?qū)С龇?wù)時，首先會對這些配置進行檢查和處理，以保證配置的正確性。配置解析邏輯封裝在 ReferenceConfig 的 init 方法中，下面進行分析。
private void init() {
// 避免重復(fù)初始化
if (initialized) {
return;
}
initialized = true;
// 檢測接口名合法性
if (interfaceName == null || interfaceName.length() == 0) {
throw new IllegalStateException("interface not allow null!");
}
// 檢測 consumer 變量是否為空，為空則創(chuàng)建
checkDefault();
appendProperties(this);
if (getGeneric() == null && getConsumer() != null) {
// 設(shè)置 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 ?------------------------------
// 從系統(tǒng)變量中獲取與接口名對應(yīng)的屬性值
String resolve = System.getProperty(interfaceName);
String resolveFile = null;
if (resolve == null || resolve.length() == 0) {
// 從系統(tǒng)屬性中獲取解析文件路徑
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);
}
}
// 獲取與接口名對應(yīng)的配置
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、協(xié)議版本信息、時間戳和進程號等信息到 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()));
}
// 非泛化服務(wù)
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)) {
// 添加重試次數(shù)配置 methodName.retries
map.put(method.getName() + ".retries", "0");
}
}
// 添加 MethodConfig 中的“屬性”字段到 attributes
// 比如 onreturn、onthrow、oninvoke 等
appendAttributes(attributes, method, prefix + "." + method.getName());
checkAndConvertImplicitConfig(method, map, attributes);
}
}
// -------------------------------? 分割線5 ?------------------------------
// 獲取服務(wù)消費者 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 到系統(tǒng)上下文中
StaticContext.getSystemContext().putAll(attributes);
// 創(chuàng)建代理類
ref = createProxy(map);
// 根據(jù)服務(wù)名，ReferenceConfig，代理類構(gòu)建 ConsumerModel，
// 并將 ConsumerModel 存入到 ApplicationModel 中
ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
}
上面的代碼很長，做的事情比較多。這里根據(jù)代碼邏輯，對代碼進行了分塊，下面我們一起來看一下。
首先是方法開始到分割線1之間的代碼。這段代碼主要用于檢測 ConsumerConfig 實例是否存在，如不存在則創(chuàng)建一個新的實例，然后通過系統(tǒng)變量或 dubbo.properties 配置文件填充 ConsumerConfig 的字段。接著是檢測泛化配置，并根據(jù)配置設(shè)置 interfaceClass 的值。
接著來看分割線1到分割線2之間的邏輯。這段邏輯用于從系統(tǒng)屬性或配置文件中加載與接口名相對應(yīng)的配置，并將解析結(jié)果賦值給 url 字段。url 字段的作用一般是用于點對點調(diào)用。
繼續(xù)向下看，分割線2和分割線3之間的代碼用于檢測幾個核心配置類是否為空，為空則嘗試從其他配置類中獲取。分割線3與分割線4之間的代碼主要用于收集各種配置，并將配置存儲到 map 中。
分割線4和分割線5之間的代碼用于處理 MethodConfig 實例。該實例包含了事件通知配置，比如 onreturn、onthrow、oninvoke 等。
分割線5到方法結(jié)尾的代碼主要用于解析服務(wù)消費者 ip，以及調(diào)用 createProxy 創(chuàng)建代理對象。關(guān)于該方法的詳細分析，將會在接下來的章節(jié)中展開。
3.2 引用服務(wù)
本節(jié)我們要從 createProxy 開始看起。從字面意思上來看，createProxy 似乎只是用于創(chuàng)建代理對象的。但實際上并非如此，該方法還會調(diào)用其他方法構(gòu)建以及合并 Invoker 實例。具體細節(jié)如下。
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;
// 根據(jù) url 的協(xié)議、scope 以及 injvm 等參數(shù)檢測是否需要本地引用
// 比如如果用戶顯式配置了 scope=local，此時 isInjvmRefer 返回 true
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
isJvmRefer = true;
} else {
isJvmRefer = false;
}
} else {
// 獲取 injvm 配置值
isJvmRefer = isInjvm().booleanValue();
}
// 本地引用
if (isJvmRefer) {
// 生成本地引用 URL，協(xié)議為 injvm
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
// 調(diào)用 refer 方法構(gòu)建 InjvmInvoker 實例
invoker = refprotocol.refer(interfaceClass, url);
// 遠程引用
} else {
// url 不為空，表明用戶可能想進行點對點調(diào)用
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) {
// 設(shè)置接口全限定名為 url 路徑
url = url.setPath(interfaceName);
}
// 檢測 url 協(xié)議是否為 registry，若是，表明用戶想使用指定的注冊中心
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
// 將 map 轉(zhuǎn)換為查詢字符串，并作為 refer 參數(shù)的值添加到 url 中
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else {
// 合并 url，移除服務(wù)提供者的一些配置（這些配置來源于用戶配置的 url 屬性），
// 比如線程池相關(guān)配置。并保留服務(wù)提供者的部分配置，比如版本，group，時間戳等
// 最后將合并后的配置設(shè)置為 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 參數(shù)到 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...");
}
}
// 單個注冊中心或服務(wù)提供者(服務(wù)直連，下同)
if (urls.size() == 1) {
// 調(diào)用 RegistryProtocol 的 refer 構(gòu)建 Invoker 實例
invoker = refprotocol.refer(interfaceClass, urls.get(0));
// 多個注冊中心或多個服務(wù)提供者，或者兩者混合
} else {
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
// 獲取所有的 Invoker
for (URL url : urls) {
// 通過 refprotocol 調(diào)用 refer 構(gòu)建 Invoker，refprotocol 會在運行時
// 根據(jù) url 協(xié)議頭加載指定的 Protocol 實例，并調(diào)用實例的 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);
// 創(chuàng)建 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);
}
上面代碼很多，不過邏輯比較清晰。
首先根據(jù)配置檢查是否為本地調(diào)用，
若是，則調(diào)用 InjvmProtocol 的 refer 方法生成 InjvmInvoker 實例。
若不是，則讀取直連配置項，或注冊中心 url，并將讀取到的 url 存儲到 urls 中。然后根據(jù) urls 元素數(shù)量進行后續(xù)操作。
若 urls 元素數(shù)量為1，則直接通過 Protocol 自適應(yīng)拓展類構(gòu)建 Invoker 實例接口。
若 urls 元素數(shù)量大于1，即存在多個注冊中心或服務(wù)直連 url，此時先根據(jù) url 構(gòu)建 Invoker。
然后再通過 Cluster 合并多個 Invoker，最后調(diào)用 ProxyFactory 生成代理類。Invoker 的構(gòu)建過程以及代理類的過程比較重要，因此接下來將分兩小節(jié)對這兩個過程進行分析。
3.2.1 創(chuàng)建 Invoker
Invoker 是 Dubbo 的核心模型，代表一個可執(zhí)行體。在服務(wù)提供方，Invoker 用于調(diào)用服務(wù)提供類。在服務(wù)消費方，Invoker 用于執(zhí)行遠程調(diào)用。Invoker 是由 Protocol 實現(xiàn)類構(gòu)建而來。Protocol 實現(xiàn)類有很多，本節(jié)會分析最常用的兩個，分別是 RegistryProtocol 和 DubboProtocol，其他的大家自行分析。下面先來分析 DubboProtocol 的 refer 方法源碼。如下：
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
optimizeSerialization(url);
// 創(chuàng)建 DubboInvoker
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
上面方法看起來比較簡單，不過這里有一個調(diào)用需要我們注意一下，即 getClients。這個方法用于獲取客戶端實例，實例類型為 ExchangeClient。ExchangeClient 實際上并不具備通信能力，它需要基于更底層的客戶端實例進行通信。比如 NettyClient、MinaClient 等，默認情況下，Dubbo 使用 NettyClient 進行通信。接下來，我們簡單看一下 getClients 方法的邏輯。
private ExchangeClient[] getClients(URL url) {
// 是否共享連接
boolean service_share_connect = false;
// 獲取連接數(shù)，默認為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;
}
這里根據(jù) connections 數(shù)量決定是獲取共享客戶端還是創(chuàng)建新的客戶端實例，默認情況下，使用共享客戶端實例。getSharedClient 方法中也會調(diào)用 initClient 方法
private ExchangeClient getSharedClient(URL url) {
String key = url.getAddress();
// 獲取帶有“引用計數(shù)”功能的 ExchangeClient
ReferenceCountExchangeClient client = referenceClientMap.get(key);
if (client != null) {
if (!client.isClosed()) {
// 增加引用計數(shù)
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);
}
// 創(chuàng)建 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 方法創(chuàng)建新的 ExchangeClient 實例，并將該實例傳給 ReferenceCountExchangeClient 構(gòu)造方法創(chuàng)建一個帶有引用計數(shù)功能的 ExchangeClient 實例。ReferenceCountExchangeClient 內(nèi)部實現(xiàn)比較簡單，就不分析了。下面我們再來看一下 initClient 方法的代碼。
private ExchangeClient initClient(URL url) {
// 獲取客戶端類型，默認為 netty
String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
// 添加編解碼和心跳包參數(shù)到 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 配置，并根據(jù)配置值決定創(chuàng)建的客戶端類型
if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
// 創(chuàng)建懶加載 ExchangeClient 實例
client = new LazyConnectExchangeClient(url, requestHandler);
} else {
// 創(chuàng)建普通 ExchangeClient 實例
client = Exchangers.connect(url, requestHandler);
}
} catch (RemotingException e) {
throw new RpcException("Fail to create remoting client for service...");
}
return client;
}
initClient 方法首先獲取用戶配置的客戶端類型，默認為 netty。然后檢測用戶配置的客戶端類型是否存在，不存在則拋出異常。最后根據(jù) lazy 配置決定創(chuàng)建什么類型的客戶端。這里的 LazyConnectExchangeClient 代碼并不是很復(fù)雜，該類會在 request 方法被調(diào)用時通過 Exchangers 的 connect 方法創(chuàng)建 ExchangeClient 客戶端，該類的代碼本節(jié)就不分析了。下面我們分析一下 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 的實現(xiàn)。
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
// 這里包含了多個調(diào)用，分別如下：
// 1. 創(chuàng)建 HeaderExchangeHandler 對象
// 2. 創(chuàng)建 DecodeHandler 對象
// 3. 通過 Transporters 構(gòu)建 Client 實例
// 4. 創(chuàng)建 HeaderExchangeClient 對象
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
}
這里的調(diào)用比較多，我們這里重點看一下 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 數(shù)量大于1，則創(chuàng)建一個 ChannelHandler 分發(fā)器
handler = new ChannelHandlerDispatcher(handlers);
}
// 獲取 Transporter 自適應(yīng)拓展類，并調(diào)用 connect 方法生成 Client 實例
return getTransporter().connect(url, handler);
}
如上，getTransporter 方法返回的是自適應(yīng)拓展類，該類會在運行時根據(jù)客戶端類型加載指定的 Transporter 實現(xiàn)類。若用戶未配置客戶端類型，則默認加載 NettyTransporter，并調(diào)用該類的 connect 方法。如下：
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
// 創(chuàng)建 NettyClient 對象
return new NettyClient(url, listener);
}
到這里就不繼續(xù)跟下去了，在往下就是通過 Netty 提供的 API 構(gòu)建 Netty 客戶端了，大家有興趣自己看看。到這里，關(guān)于 DubboProtocol 的 refer 方法就分析完了。接下來，繼續(xù)分析 RegistryProtocol 的 refer 方法邏輯。
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 取 registry 參數(shù)值，并將其設(shè)置為協(xié)議頭
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);
}
// 將 url 查詢字符串轉(zhuǎn)為 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 實例，并調(diào)用 doRefer 繼續(xù)執(zhí)行服務(wù)引用邏輯
return doRefer(getMergeableCluster(), registry, type, url);
}
}
// 調(diào)用 doRefer 繼續(xù)執(zhí)行服務(wù)引用邏輯
return doRefer(cluster, registry, type, url);
}
上面代碼首先為 url 設(shè)置協(xié)議頭，然后根據(jù) url 參數(shù)加載注冊中心實例。然后獲取 group 配置，根據(jù) group 配置決定 doRefer 第一個參數(shù)的類型。這里的重點是 doRefer 方法，如下：
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// 創(chuàng)建 RegistryDirectory 實例
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
// 設(shè)置注冊中心和協(xié)議
directory.setRegistry(registry);
directory.setProtocol(protocol);
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
// 生成服務(wù)消費者鏈接
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
// 注冊服務(wù)消費者，在 consumers 目錄下新節(jié)點
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 等節(jié)點數(shù)據(jù)
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
// 一個注冊中心可能有多個服務(wù)提供者，因此這里需要將多個服務(wù)提供者合并為一個
Invoker invoker = cluster.join(directory);
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
return invoker;
}
如上，doRefer 方法創(chuàng)建一個 RegistryDirectory 實例，然后生成服務(wù)者消費者鏈接，并向注冊中心進行注冊。注冊完畢后，緊接著訂閱 providers、configurators、routers 等節(jié)點下的數(shù)據(jù)。完成訂閱后，RegistryDirectory 會收到這幾個節(jié)點下的子節(jié)點信息。由于一個服務(wù)可能部署在多臺服務(wù)器上，這樣就會在 providers 產(chǎn)生多個節(jié)點，這個時候就需要 Cluster 將多個服務(wù)節(jié)點合并為一個，并生成一個 Invoker。關(guān)于 RegistryDirectory 和 Cluster，本文不打算進行分析，相關(guān)分析將會在隨后的文章中展開。
3.2.2 創(chuàng)建代理
Invoker 創(chuàng)建完畢后，接下來要做的事情是為服務(wù)接口生成代理對象。有了代理對象，即可進行遠程調(diào)用。代理對象生成的入口方法為 ProxyFactory 的 getProxy，接下來進行分析。
public <T> T getProxy(Invoker<T> invoker) throws RpcException {
// 調(diào)用重載方法
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];
// 設(shè)置服務(wù)接口類和 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 協(xié)議提供泛化調(diào)用支持，參考 pull request #1827
if (!invoker.getInterface().equals(GenericService.class) && generic) {
int len = interfaces.length;
Class<?>[] temp = interfaces;
// 創(chuàng)建新的 interfaces 數(shù)組
interfaces = new Class<?>[len + 1];
System.arraycopy(temp, 0, interfaces, 0, len);
// 設(shè)置 GenericService.class 到數(shù)組中
interfaces[len] = GenericService.class;
}
// 調(diào)用重載方法
return getProxy(invoker, interfaces);
}
public abstract <T> T getProxy(Invoker<T> invoker, Class<?>[] types);
如上，上面大段代碼都是用來獲取 interfaces 數(shù)組的，我們繼續(xù)往下看。getProxy(Invoker, Class<?>[]) 這個方法是一個抽象方法，下面我們到 JavassistProxyFactory 類中看一下該方法的實現(xiàn)代碼。
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
// 生成 Proxy 子類（Proxy 是抽象類）。并調(diào)用 Proxy 子類的 newInstance 方法創(chuàng)建 Proxy 實例
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
上面代碼并不多，首先是通過 Proxy 的 getProxy 方法獲取 Proxy 子類，然后創(chuàng)建 InvokerInvocationHandler 對象，并將該對象傳給 newInstance 生成 Proxy 實例。InvokerInvocationHandler 實現(xiàn)自 JDK 的 InvocationHandler 接口，具體的用途是攔截接口類調(diào)用。該類邏輯比較簡單，這里就不分析了。下面我們重點關(guān)注一下 Proxy 的 getProxy 方法，如下。
public static Proxy getProxy(Class<?>... ics) {
// 調(diào)用重載方法
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;
}
}
// 并發(fā)控制，保證只有一個線程可以進行后續(xù)操作
if (value == PendingGenerationMarker) {
try {
// 其他線程在此處進行等待
cache.wait();
} catch (InterruptedException e) {
}
} else {
// 放置標志位到緩存中，并跳出 while 循環(huán)進行后續(xù)操作
cache.put(key, PendingGenerationMarker);
break;
}
}
while (true);
}
long id = PROXY_CLASS_COUNTER.getAndIncrement();
String pkg = null;
ClassGenerator ccp = null, ccm = null;
try {
// 創(chuàng)建 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();
// 獲取參數(shù)列表
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 方法調(diào)用語句，如下：
// 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);
// 添加方法名、訪問控制符、參數(shù)列表、方法代碼等信息到 ClassGenerator 中
ccp.addMethod(method.getName(), method.getModifiers(), rt, pts, method.getExceptionTypes(), code.toString());
}
}
if (pkg == null)
pkg = PACKAGE_NAME;
// 構(gòu)建接口代理類名稱：pkg + ".proxy" + id，比如 org.apache.dubbo.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 參數(shù)的構(gòu)造方法，比如：
// porxy0(java.lang.reflect.InvocationHandler arg0) {
// handler=$1;
// }
ccp.addConstructor(Modifier.PUBLIC, new Class<?>[]{InvocationHandler.class}, new Class<?>[0], "handler=$1;");
// 為接口代理類添加默認構(gòu)造方法
ccp.addDefaultConstructor();
// 生成接口代理類
Class<?> clazz = ccp.toClass();
clazz.getField("methods").set(null, methods.toArray(new Method[0]));
// 構(gòu)建 Proxy 子類名稱，比如 Proxy1，Proxy2 等
String fcn = Proxy.class.getName() + id;
ccm = ClassGenerator.newInstance(cl);
ccm.setClassName(fcn);
ccm.addDefaultConstructor();
ccm.setSuperClass(Proxy.class);
// 為 Proxy 的抽象方法 newInstance 生成實現(xiàn)代碼，形如：
// public Object newInstance(java.lang.reflect.InvocationHandler h) {
// return new org.apache.dubbo.proxy0($1);
// }
ccm.addMethod("public Object newInstance(" + InvocationHandler.class.getName() + " h){ return new " + pcn + "($1); }");
// 生成 Proxy 實現(xiàn)類
Class<?> pc = ccm.toClass();
// 通過反射創(chuàng)建 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;
}
上面代碼比較復(fù)雜，我們寫了大量的注釋。大家在閱讀這段代碼時，要搞清楚 ccp 和 ccm 的用途，不然會被搞暈。ccp 用于為服務(wù)接口生成代理類，比如我們有一個 DemoService 接口，這個接口代理類就是由 ccp 生成的。ccm 則是用于為 org.apache.dubbo.common.bytecode.Proxy 抽象類生成子類，主要是實現(xiàn) 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;
}
}
到這里代理類生成邏輯就分析完了。整個過程比較復(fù)雜，大家需要耐心看一下。
4.總結(jié)
本篇文章對服務(wù)引用的過程進行了較為詳盡的分析，還有一些邏輯暫時沒有分析到，比如 Directory、Cluster。這些接口及實現(xiàn)類功能比較獨立，后續(xù)會單獨成文進行分析。暫時我們可以先把這些類看成黑盒，只要知道這些類的用途即可。關(guān)于服務(wù)引用過程就分析到這里。
補充
服務(wù)引用調(diào)用鏈
ReferenceBean.getObject()
-->ReferenceConfig.get() //在ReferenceConfig中初始化了三個變量Protocol refprotocol，Cluster cluster，ProxyFactory proxyFactory
-->ReferenceConfig.init() //配置檢查
-->createProxy(map); //創(chuàng)建代理proxy
-->refprotocol.refer(interfaceClass, urls.get(0)) //生成invoker
-->ExtensionLoader.getExtensionLoader(Protocol.class).getExtension("registry")
-->extension.refer(arg0, arg1)
-->ProtocolFilterWrapper.refer
-->RegistryProtocol.refer(type,url) //refprotocol.refer最終執(zhí)行RegistryProtocol.refer方法
-->registryFactory.getRegistry(url);//和zk建立連接
-->doRefer(cluster, registry, type, url)
-->registry.register //創(chuàng)建消費者zk節(jié)點,節(jié)點名為：dubbo/com.alibaba.dubbo.demo.DemoService/consumers
-->directory.subscribe //訂閱zk節(jié)點，包括providers，configurators，routers
-->cluster.join(directory) //多個invoker偽裝成cluster invoker，到此invoker創(chuàng)建完成
-->proxyFactory.getProxy(invoker) //生成proxy
簡單歸納為這幾步：
ReferenceBean.getObject() --> ReferenceConfig.init() -->RegistryProtocol.doRefer--> cluster.join(directory) --> proxyFactory.getProxy(invoker)
