Dubbo系列之 （五）服務訂閱（2）
輔助連結
Dubbo系列之 （一）SPI擴充套件
Dubbo系列之 （二）Registry註冊中心-註冊（1）
Dubbo系列之 （三）Registry註冊中心-註冊（2）
Dubbo系列之 （四）服務訂閱（1）
Dubbo系列之 （五）服務訂閱（2）
服務訂閱，閱讀程式碼前的一些思考？
思考的過程和設計思想如下：
1、我們想要進行遠端服務的呼叫，那麼肯定要建立網路連線，不妨改用TCP長連線，並設計通訊協議，並封裝為一個類，不妨叫做ExchangeClient。用它來進行網路通訊。
2、有了可以進行遠端通訊的服務物件ExchangeClient後，我們可以把遠端服務封裝為一個Invoker物件，這個Invoker物件內部採用自已定義的協議與遠端伺服器通訊，不妨叫做DubboInvoker，因為採用了dubbo協議來進行網路通訊的。
3、有了這個DubboInvoker 我就可以根據dubbo協議與遠端服務通訊了，但是我還想在本地增加一些過濾器Filter,或者監聽器Listener。沒關係，直接通過責任鏈模式，把這些Filter與這個DubboInvoker進行連結。返回的一個ProtocolFilterWrapper物件。
4、同理，如果需要一些監聽器的功能怎麼辦，同樣進行一次封裝。把ProtocolFilterWraper封裝到Listener型別的Invoker物件，不妨叫做ListenerInvokerWrapper。
5、現在考慮遠端服務提供者有很多個，那麼我對每個遠端服務都需要有一個ListenerInvokerWrapper的物件。如下：
Demoservice::196.254.324.1 ListenerInvokerWrapper1
Demoservice::196.254.324.2 ListenerInvokerWrapper2
Demoservice::196.254.324.3 ListenerInvokerWrapper3
Demoservice::196.254.324.4 ListenerInvokerWrapper4
Demoservice::196.254.324.5 ListenerInvokerWrapper5
.....
6、服務太多了，在本地這樣建立太費事了。引入了註冊中心，直接把服務註冊到服務中心上，然後客戶端直接從註冊中心拉取。我們把拉取到的服務，統稱為服務目錄。並且它是從註冊中心拉取到的，那麼不妨名字就叫做RegistryDirectory。那麼這個服務目錄裡肯定包含了上面的遠端服務呼叫物件ListenerInvokerWrapper。我們把這些物件放到服務目錄的成員上，名字就叫做urlInvokerMap。key: Demoservice::xxxx。value:ListenerInvokerWrapper。
7、現在我們可以在本地呼叫RegistryDirectory物件，與遠端服務通訊了，想調哪個服務就從urlInvokerMap取出一個進行呼叫即可。但是每次指定一個遠端伺服器，不僅太麻煩了，而且也會造成流量不均勻，負載不平衡。那麼我們就通過通過負載均衡策略來選擇一個服務呼叫。就取名LoadBalance吧。他有個方法select。入參就是我們的服務目錄RegistryDirectory。那麼通過LoadBalance.select(RegistryDirectory) 得到一個我們想要的通訊的遠端服務即可。目前負載均衡演算法有一致性Hash演算法，隨機演算法、權重輪訓演算法、最短響應時間演算法、最少活躍數演算法。
8、有了負載均衡演算法LoadBalance後，我想要這樣的功能，當服務呼叫失敗的時候，我可以重試，或者直接直接失敗。那我就把有這種能力服務呼叫，稱為一個叢集Cluster。他有一個方法叫做join。入參還是服務目錄RegistryDirectory。返回一個具有快速失敗、或者重試的服務呼叫，不妨叫AbstractClusterInvoker。每個不同的策略都去實現它。並且這個物件內部通過LoadBalance來選擇一個服務進行呼叫，失敗後的策略（是否重試或失敗）由我決定。
9、目前我們已經有了一個XXXclusterInvoker 物件，它具有快速失敗或者重試等功能，且具有負載均衡演算法的遠端服務呼叫物件。但是有時，這些遠端服務提供者這的qps不達標，或者新上線的服務有問題，或者遠端服務呼叫失敗後，可以在本地模擬的呼叫，返回一個mock物件。那麼我們重新對XXXclusterInvoker進行封裝，就命名為MockClusterInvoker,具有Mock功能，且具有叢集能力。它持有我們的服務目錄RegistryDirectory和XXXclusterInvoker物件。
10、目前我們已經有了一個MockClusterInvoker物件。但是這個invoker物件和我們像本地一樣呼叫服務還是有點差別，最後我們直接通過Java的動態代理計算Proxy.newInstance()來建立一個具體的服務物件DemoService,並且在InvokeHandler內部呼叫我們的MockClusterInvoker物件的invoke 方法。
11、比如我們的DubboInvoker是通過Java 非同步執行緒CompletableFuture實現的話，如果需要轉為同步，還可以對其封裝從非同步轉為同步的Invoker，不妨命名為AsyncToSyncInvoker。
則最終在服務消費端呈現給我們如下一個遠端服務代理物件。
ReferenceBean#getObject()
在上一章節，已經說明了getObject()物件的呼叫時機，內部呼叫的ReferenceConfig#init方法，該init()方法主要做了如下幾件事情：
1、預設的配置進行填充，比如registry,application等屬性。
2、校驗配置是否填寫正確，比如<dubbo:reference />中的stub 和mock 是否配置，配置了是否正確。
3、通過SPI機制獲取Protocol$Adaptive自適應協議，通過Protocol$Adaptive#refer()方法得到一個MockClusterInvoker物件。該方法的呼叫內容基本和上面的猜想設計一致。
1）和註冊中心建立tcp連線。
2）把當前的訂閱服務註冊到註冊中心上的consumer節點上。
3）從註冊中心中把訂閱的服務列表拉取到本地，即RegistryDirectory。
4）根據上面類似猜想建立MockClusterInvoker返回。
4、通過SPI機制獲取ProxyFactory$Adaptive自適應代理工廠，然後通過這個代理工廠建立動態代理物件,並把這個代理物件賦值給ref屬性。
REF_PROTOCOL.refer(interfacere,registryUrl)
服務的訂閱核心就是這條語句，這條語句博大精深。僅僅一條語句把所有的訂閱工作完成了。
1、首先根據SPI機制獲取自適應的協議物件。語句如下：
ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
該語句建立了Protocol$Apdative。它有個自適應refer方法如下：
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
if (type == null)
throw new IllegalArgumentException("url == null");
String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
if (extName == null)
throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.Protocol) name from url (" + url.toString() + ") use keys([protocol])");
Protocol extension = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(extName);
return extension.refer(type, url);
}
2、Protocol$Apdative#refer()方法內部又通過引數的url的協議頭和SPI機制獲取一個具體的協議。顯而易見，url.getProtocol()返回的是registry。因為當前是服務訂閱。所以是registry打頭。那麼返回的Protocol具體型別就是RegistryProtocol。但是Protocol擴充套件點有包裹型別：ProtocolListenerWrapper、ProtocolFilterWrapper。所以最終返回的是ProtocolListenerWrapper型別的協議。檢視這個2個包裹型別的refer()方法：
類ProtocolListenerWrapper
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
if (UrlUtils.isRegistry(url)) {
return protocol.refer(type, url);
}
return new ListenerInvokerWrapper<T>(protocol.refer(type, url),
Collections.unmodifiableList(
ExtensionLoader.getExtensionLoader(InvokerListener.class)
.getActivateExtension(url, INVOKER_LISTENER_KEY)));
}
類ProtocolFilterWrapper
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
if (UrlUtils.isRegistry(url)) {
return protocol.refer(type, url);
}
return buildInvokerChain(protocol.refer(type, url), REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
}
3、所以Protocol$Apdative#refer()內部的getExtension返回的是ProtocolListenerWrapper的Protocol。又因為url是註冊url，所以滿足UrlUtils.isRegistry(url)==true.直接進行一次傳遞呼叫。
4、最終調到RegistryProtocol#refer()。程式碼如下：
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
url = getRegistryUrl(url);
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(REFER_KEY));
String group = qs.get(GROUP_KEY);
if (group != null && group.length() > 0) {
if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
return doRefer(getMergeableCluster(), registry, type, url);
}
}
return doRefer(cluster, registry, type, url);
}
即得到註冊中心Registry，一般是ZookeeperRegistry。獲取註冊中心的內容在之前的章節已見過，就不在多說了。接著會呼叫doRefer()方法。
在看doRefer()方法之前，我們來看下其定義：
Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url)；
出參：
返回值就是我們需要的Invoker物件。
入參：
cluster：叢集物件Cluster$Adaptive，通過Spi獲取.內部getExtension獲取Cluster,預設為FailoverCluster。
registry：註冊中心
type:訂閱的介面型別
url:服務註冊連結註冊中心URL。
5、Cluster的join 介面如下：
Cluster$Adaptive#join()內部實際是預設呼叫的是FailoverCluster#join()。
並且Cluster擴充套件點也有其Wrapper類，即MockClusterWrapper。所以Cluster$Adaptive#join()的方法呼叫
Cluster extension = ExtensionLoader.getExtensionLoader(Cluster.class).getExtension(extName);
返回的extension是MockClusterWrapper，MockClusterWrapper#join()程式碼如下：
return new MockClusterInvoker<T>(directory,
this.cluster.join(directory));
}
所以Cluster$Adaptive#join()返回的Invoker型別是MockClusterInvoker。MockClusterWrapper持有的cluster是FailoverCluster,所以MockClusterInvoker內部持有invoker型別是FailoverClusterInvoker。
6、原始碼doRefer()
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// new 一個服務目錄，訂閱服務型別為type 的 RegistryDirectory
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
// 設定註冊中心
directory.setRegistry(registry);
//設定協議，即Protocol$Adaptive
directory.setProtocol(protocol);
// all attributes of REFER_KEY
//獲取訂閱引數
Map<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());
//構建訂閱URL ,以consumer//打頭
URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
//把該url註冊到註冊中心上
if (directory.isShouldRegister()) {
directory.setRegisteredConsumerUrl(subscribeUrl);
registry.register(directory.getRegisteredConsumerUrl());
}
//設定路由鏈
directory.buildRouterChain(subscribeUrl);
//重點，重中之重。這裡訂閱服務，並且會拉取遠端服務invoker 到directory物件的urlInvokerMap成員中。
directory.subscribe(toSubscribeUrl(subscribeUrl));
//由上面分析，得到是MockClusterInvoker
Invoker<T> invoker = cluster.join(directory);
//查詢註冊協議監聽器，沒有設定為空
List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
if (CollectionUtils.isEmpty(listeners)) {
return invoker;
}
// 如果有其監聽器進行監聽器onRefer()呼叫，並返回RegistryInvokerWrapper包裹型別。
RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker, subscribeUrl);
for (RegistryProtocolListener listener : listeners) {
listener.onRefer(this, registryInvokerWrapper);
}
return registryInvokerWrapper;
}
本地動態代理物件建立createProxy()
/**
* 核心，通過配置的元資訊，建立一個代理物件
* @param map
* @return
*/
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
private T createProxy(Map<String, String> map) {
// 首先判斷本地是否有Service提供者，
if (shouldJvmRefer(map)) {
//如果有，匯出jvm匯出refer
URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
invoker = REF_PROTOCOL.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
} else {
urls.clear();
//指定服務提供者URL。點對點比如在<dubbo:reference url="dubbo://xxxxx:12222">
if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (StringUtils.isEmpty(url.getPath())) {
url = url.setPath(interfaceName);
}
if (UrlUtils.isRegistry(url)) {
urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
} else {
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else { // assemble URL from register center's configuration
// if protocols not injvm checkRegistry
//如果不是jvm 協議，一般是dubbo
if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
//檢測註冊中心
checkRegistry();
//根據註冊中心地址，得到註冊服務
//registry://106.52.187.48:2181/org.apache.dubbo.registry.RegistryService
// ?application=dubbo-demo-annotation-consumer&dubbo=2.0.2&pid=9757®istry=zookeeper×tamp=1597380362736
List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
if (CollectionUtils.isNotEmpty(us)) {
for (URL u : us) {
//對每個註冊中心URL，得到監控URL
URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
if (monitorUrl != null) {
map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
}
}
if (urls.isEmpty()) {
throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
}
}
}
//如果註冊中心之一一個的話，一般就一個註冊中心
if (urls.size() == 1) {
invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
} else {
//多個註冊中心時，Protocol$Adaptive
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
//把其得到的Invoker 填入invokers
invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
if (UrlUtils.isRegistry(url)) {
registryURL = url; // use last registry url
}
}
//多註冊中心，多訂閱場景
if (registryURL != null) { // registry url is available
// for multi-subscription scenario, use 'zone-aware' policy by default
URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
// The invoker wrap relation would be like: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
//通過叢集，返回一個invoker
invoker = CLUSTER.join(new StaticDirectory(u, invokers));
} else { // not a registry url, must be direct invoke.
invoker = CLUSTER.join(new StaticDirectory(invokers));
}
}
}
if (shouldCheck() && !invoker.isAvailable()) {
invoker.destroy();
throw new IllegalStateException("Failed to check the status of the service "
+ interfaceName
+ ". No provider available for the service "
+ (group == null ? "" : group + "/")
+ interfaceName +
(version == null ? "" : ":" + version)
+ " from the url "
+ invoker.getUrl()
+ " to the consumer "
+ NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
/**
* @since 2.7.0
* ServiceData Store
*/
/**
*
* 這裡是釋出元資料資訊
*/
String metadata = map.get(METADATA_KEY);
WritableMetadataService metadataService = WritableMetadataService.getExtension(metadata == null ? DEFAULT_METADATA_STORAGE_TYPE : metadata);
if (metadataService != null) {
URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
metadataService.publishServiceDefinition(consumerURL);
}
// create service proxy
//通過動態代理把invoker 轉化為具體的服務型別
return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
}
上面核心的程式碼invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0))已分析，接下下來就是通過PROXY_FACTORY.getProxy()建立活動，之後服務呼叫上進行分析。其他元資料的註冊，等之後講解配置中心時進行講解。
接下來，以一個圖解來描述服務訂閱的過程。在下一章節來描述如何具體的拉取遠端服務invoker到服務目錄RegistryDirectory上的urlInvokerMap。