輔助鏈接
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。
