摘要：是用來(lái)監(jiān)聽(tīng)處理注冊(cè)數(shù)據(jù)變更的事件。這里的是節(jié)點(diǎn)的接口，里面協(xié)定了關(guān)于節(jié)點(diǎn)的一些操作方法，我們可以來(lái)看看源代碼獲得節(jié)點(diǎn)地址判斷節(jié)點(diǎn)是否可用銷(xiāo)毀節(jié)點(diǎn)三這個(gè)接口是注冊(cè)中心的工廠接口，用來(lái)返回注冊(cè)中心的對(duì)象。
注冊(cè)中心——開(kāi)篇
目標(biāo)：解釋注冊(cè)中心在dubbo框架中作用，dubbo-registry-api源碼解讀
注冊(cè)中心是什么？
服務(wù)治理框架中可以大致分為服務(wù)通信和服務(wù)管理兩個(gè)部分，服務(wù)管理可以分為服務(wù)注冊(cè)、服務(wù)發(fā)現(xiàn)以及服務(wù)被熱加工介入，服務(wù)提供者Provider會(huì)往注冊(cè)中心注冊(cè)服務(wù)，而消費(fèi)者Consumer會(huì)從注冊(cè)中心中訂閱相關(guān)的服務(wù)，并不會(huì)訂閱全部的服務(wù)。
官方文檔給出了Provider、Consumer以及Registry之間的依賴(lài)關(guān)系：
從上圖看，可以清晰的看到Registry所起到的作用，我舉個(gè)例子，Registry類(lèi)似于一個(gè)自動(dòng)售貨機(jī)，服務(wù)提供者類(lèi)似于一個(gè)商品生產(chǎn)者，他會(huì)往這個(gè)自動(dòng)售賣(mài)機(jī)中添加商品，也就是注冊(cè)服務(wù)，而消費(fèi)者則會(huì)到注冊(cè)中心中購(gòu)買(mǎi)自己需要的商品，也就是訂閱對(duì)應(yīng)的服務(wù)。這樣解釋?xiě)?yīng)該就可以比較直觀的感受到注冊(cè)中心所擔(dān)任的是什么角色。
dubbo-registry-api的解讀
首先我們來(lái)看看這個(gè)包下的結(jié)構(gòu)：
可以很清晰的看到dubbo內(nèi)部支持的四種注冊(cè)中心實(shí)現(xiàn)方式，分別是dubbo、multicast、zookeeper、redis。他們都依賴(lài)于support包下面的類(lèi)。根據(jù)上圖的依賴(lài)關(guān)系，我會(huì)從上往下講解dubbo中對(duì)于注冊(cè)中心的設(shè)計(jì)以及實(shí)現(xiàn)。
（一）RegistryService
該接口是注冊(cè)中心模塊的服務(wù)接口，提供了注冊(cè)、取消注冊(cè)、訂閱、取消訂閱以及查詢(xún)符合條件的已注冊(cè)數(shù)據(jù)。它的源代碼我就不貼出來(lái)了，可以查看官方文檔中相關(guān)部分，還給出了中文注釋。
RegistryService源碼地址：http://dubbo.apache.org/zh-cn...
我們可以從注釋中看到各個(gè)方法要處理的契約都在上面寫(xiě)明了。這個(gè)接口就是協(xié)定了注冊(cè)中心的功能，這里統(tǒng)一說(shuō)明一下URL，又再次提到URL了，在上篇文章中就說(shuō)明了dubbo是以總線模式來(lái)時(shí)刻傳遞和保存配置信息的，也就是配置信息都被放在URL上進(jìn)行傳遞，隨時(shí)可以取得相關(guān)配置信息，而這里提到了URL有別的作用，就是作為類(lèi)似于節(jié)點(diǎn)的作用，首先服務(wù)提供者（Provider）啟動(dòng)時(shí)需要提供服務(wù)，就會(huì)向注冊(cè)中心寫(xiě)下自己的URL地址。然后消費(fèi)者啟動(dòng)時(shí)需要去訂閱該服務(wù)，則會(huì)訂閱Provider注冊(cè)的地址，并且消費(fèi)者也會(huì)寫(xiě)下自己的URL。繼續(xù)拿我上面的例子，商品生產(chǎn)者生產(chǎn)完商品，它會(huì)在把該商品放在自動(dòng)售賣(mài)機(jī)的某一個(gè)欄目?jī)?nèi)，二消費(fèi)者需要買(mǎi)該商品的時(shí)候，就是通過(guò)該地址去購(gòu)買(mǎi)，并且會(huì)留下自己的購(gòu)買(mǎi)記錄。下面來(lái)講講各個(gè)方法：
注冊(cè)，如果看懂我上面說(shuō)的url的作用，那么就很清楚該方法的作用了，這里強(qiáng)調(diào)一點(diǎn)，就是注釋中講到的允許URI相同但參數(shù)不同的URL并存，不能覆蓋，也就是說(shuō)url值必須唯一的，不能有一模一樣。
void register(URL url);
取消注冊(cè)，該方法也很簡(jiǎn)單，就是取消注冊(cè)，也就是商品生產(chǎn)者不在銷(xiāo)售該商品， 需要把東西從自動(dòng)售賣(mài)機(jī)上取下來(lái)，欄目也要取出，這里強(qiáng)調(diào)按全URL匹配取消注冊(cè)。
void unregister(URL url);
訂閱，這里不是根據(jù)全URL匹配訂閱的，而是根據(jù)條件去訂閱，也就是說(shuō)可以訂閱多個(gè)服務(wù)。listener是用來(lái)監(jiān)聽(tīng)處理注冊(cè)數(shù)據(jù)變更的事件。
void subscribe(URL url, NotifyListener listener);
取消訂閱，這是按照全URL匹配去取消訂閱的。
void unsubscribe(URL url, NotifyListener listener);
查詢(xún)注冊(cè)列表，通過(guò)url進(jìn)行條件查詢(xún)所匹配的所有URL集合。
List lookup(URL url);
（二）Registry
注冊(cè)中心接口，該接口很好理解，就是把節(jié)點(diǎn)以及注冊(cè)中心服務(wù)的方法整合在了這個(gè)接口里面。我們來(lái)看看源代碼：
public interface Registry extends Node, RegistryService {
}
可以看到該接口并沒(méi)有自己的方法，就是繼承了Node和RegistryService接口。這里的Node是節(jié)點(diǎn)的接口，里面協(xié)定了關(guān)于節(jié)點(diǎn)的一些操作方法，我們可以來(lái)看看源代碼：
public interface Node {
//獲得節(jié)點(diǎn)地址
URL getUrl();
//判斷節(jié)點(diǎn)是否可用
boolean isAvailable();
//銷(xiāo)毀節(jié)點(diǎn)
void destroy();
}
（三）RegistryFactory
這個(gè)接口是注冊(cè)中心的工廠接口，用來(lái)返回注冊(cè)中心的對(duì)象。來(lái)看看它的源碼：
@SPI("dubbo")
public interface RegistryFactory {
@Adaptive({"protocol"})
Registry getRegistry(URL url);
}
本來(lái)方法上有一些英文注釋?zhuān)瑢?xiě)的是關(guān)于連接注冊(cè)中心需處理的契約，具體的可以直接看官方文檔，還是中文的。
地址：http://dubbo.apache.org/zh-cn...
該接口是一個(gè)可擴(kuò)展接口，可以看到該接口上有個(gè)@SPI注解，并且默認(rèn)值為dubbo，也就是默認(rèn)擴(kuò)展的是DubboRegistryFactory，并且可以在getRegistry方法上可以看到有@Adaptive注解，那么該接口會(huì)動(dòng)態(tài)生成一個(gè)適配器RegistryFactory$Adaptive，并且會(huì)去首先擴(kuò)展url.protocol的值對(duì)應(yīng)的實(shí)現(xiàn)類(lèi)。關(guān)于SPI擴(kuò)展機(jī)制請(qǐng)觀看《dubbo源碼解析（二）Dubbo擴(kuò)展機(jī)制SPI》。
（四）NotifyListener
該接口只有一個(gè)notify方法，通知監(jiān)聽(tīng)器。當(dāng)收到服務(wù)變更通知時(shí)觸發(fā)。來(lái)看看它的源碼：
public interface NotifyListener {
/**
* 當(dāng)收到服務(wù)變更通知時(shí)觸發(fā)。
*
* 通知需處理契約：
* 1. 總是以服務(wù)接口和數(shù)據(jù)類(lèi)型為維度全量通知，即不會(huì)通知一個(gè)服務(wù)的同類(lèi)型的部分?jǐn)?shù)據(jù)，用戶(hù)不需要對(duì)比上一次通知結(jié)果。
* 2. 訂閱時(shí)的第一次通知，必須是一個(gè)服務(wù)的所有類(lèi)型數(shù)據(jù)的全量通知。
* 3. 中途變更時(shí)，允許不同類(lèi)型的數(shù)據(jù)分開(kāi)通知，比如：providers, consumers, routers, overrides，允許只通知其中一種類(lèi)型，但該類(lèi)型的數(shù)據(jù)必須是全量的，不是增量的。
* 4. 如果一種類(lèi)型的數(shù)據(jù)為空，需通知一個(gè)empty協(xié)議并帶category參數(shù)的標(biāo)識(shí)性URL數(shù)據(jù)。
* 5. 通知者(即注冊(cè)中心實(shí)現(xiàn))需保證通知的順序，比如：?jiǎn)尉€程推送，隊(duì)列串行化，帶版本對(duì)比。
*
* @param urls 已注冊(cè)信息列表，總不為空，含義同{@link com.alibaba.dubbo.registry.RegistryService#lookup(URL)}的返回值。
*/
void notify(List urls);
}
（五）support包下的AbstractRegistry
AbstractRegistry實(shí)現(xiàn)的是Registry接口，是Registry的抽象類(lèi)。為了減輕注冊(cè)中心的壓力，在該類(lèi)中實(shí)現(xiàn)了把本地URL緩存到property文件中的機(jī)制，并且實(shí)現(xiàn)了注冊(cè)中心的注冊(cè)、訂閱等方法。
源碼注釋地址：https://github.com/CrazyHZM/i...
1.屬性
// URL的地址分隔符，在緩存文件中使用，服務(wù)提供者的URL分隔
private static final char URL_SEPARATOR = " ";
// URL地址分隔正則表達(dá)式，用于解析文件緩存中服務(wù)提供者URL列表
private static final String URL_SPLIT = "s+";
// 日志輸出
protected final Logger logger = LoggerFactory.getLogger(getClass());
// 本地磁盤(pán)緩存，有一個(gè)特殊的key值為registies，記錄的是注冊(cè)中心列表，其他記錄的都是服務(wù)提供者列表
private final Properties properties = new Properties();
// 緩存寫(xiě)入執(zhí)行器
private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));
// 是否同步保存文件標(biāo)志
private final boolean syncSaveFile;
//數(shù)據(jù)版本號(hào)
private final AtomicLong lastCacheChanged = new AtomicLong();
// 已注冊(cè) URL 集合
// 注冊(cè)的 URL 不僅僅可以是服務(wù)提供者的，也可以是服務(wù)消費(fèi)者的
private final Set registered = new ConcurrentHashSet();
// 訂閱URL的監(jiān)聽(tīng)器集合
private final ConcurrentMap> subscribed = new ConcurrentHashMap>();
// 某個(gè)消費(fèi)者被通知的某一類(lèi)型的 URL 集合
// 第一個(gè)key是消費(fèi)者的URL，對(duì)應(yīng)的就是哪個(gè)消費(fèi)者。
// value是一個(gè)map集合，該map集合的key是分類(lèi)的意思，例如providers、routes等，value就是被通知的URL集合
private final ConcurrentMap>> notified = new ConcurrentHashMap>>();
// 注冊(cè)中心 URL
private URL registryUrl;
// 本地磁盤(pán)緩存文件，緩存注冊(cè)中心的數(shù)據(jù)
private File file;
理解屬性的含義對(duì)于后面去解讀方法很有幫助，從上面可以看到除了注冊(cè)中心相關(guān)的一些屬性外，可以看到好幾個(gè)是個(gè)屬性跟磁盤(pán)緩存文件和讀寫(xiě)文件有關(guān)的，這就是上面提到的把URL緩存到本地property的相關(guān)屬性這里有幾個(gè)需要關(guān)注的點(diǎn)：
properties：properties的數(shù)據(jù)跟本地文件的數(shù)據(jù)同步，當(dāng)啟動(dòng)時(shí)，會(huì)從文件中讀取數(shù)據(jù)到properties，而當(dāng)properties中數(shù)據(jù)變化時(shí)，會(huì)寫(xiě)入到file。而properties是一個(gè)key對(duì)應(yīng)一個(gè)列表，比如說(shuō)key就是消費(fèi)者的url，而值就是服務(wù)提供者列表、路由規(guī)則列表、配置規(guī)則列表。就是類(lèi)似屬性notified的含義。需要注意的是properties有一個(gè)特殊的key為registies，記錄的是注冊(cè)中心列表。
lastCacheChanged：因?yàn)槊看螌?xiě)入file都是全部覆蓋的寫(xiě)入，不是增量的去寫(xiě)入到文件，所以需要有這個(gè)版本號(hào)來(lái)避免老版本覆蓋新版本。
notified：跟properties的區(qū)別是第一數(shù)據(jù)來(lái)源不是文件，而是從注冊(cè)中心中讀取，第二個(gè)notified根據(jù)分類(lèi)把同一類(lèi)的值做了聚合。
2.構(gòu)造方法AbstractRegistry
先來(lái)看看源碼：
public AbstractRegistry(URL url) {
// 把url放到registryUrl中
setUrl(url);
// Start file save timer
// 從url中讀取是否同步保存文件的配置，如果沒(méi)有值默認(rèn)用異步保存文件
syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
// 獲得file路徑
String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getParameter(Constants.APPLICATION_KEY) + "-" + url.getAddress() + ".cache");
File file = null;
if (ConfigUtils.isNotEmpty(filename)) {
//創(chuàng)建文件
file = new File(filename);
if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) {
if (!file.getParentFile().mkdirs()) {
throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
}
}
}
this.file = file;
// 把文件里面的數(shù)據(jù)寫(xiě)入properties
loadProperties();
// 通知監(jiān)聽(tīng)器，URL 變化結(jié)果
notify(url.getBackupUrls());
}
需要關(guān)注的幾個(gè)點(diǎn)：
比如是否同步保存文件、比如保存的文件路徑都優(yōu)先選擇URL上的配置，如果沒(méi)有相關(guān)的配置，再選用默認(rèn)配置。
構(gòu)造AbstractRegistry會(huì)有把文件里面的數(shù)據(jù)寫(xiě)入到properties的操作以及通知監(jiān)聽(tīng)器url變化結(jié)果，相關(guān)方法介紹在下面給出。
3.filterEmpty
protected static List filterEmpty(URL url, List urls) {
if (urls == null || urls.isEmpty()) {
List result = new ArrayList(1);
result.add(url.setProtocol(Constants.EMPTY_PROTOCOL));
return result;
}
return urls;
}
這個(gè)方法的源碼都不需要解釋了，很簡(jiǎn)單，就是判斷url集合是否為空，如果為空，則把url中key為empty的值加入到集合。該方法只有在notify方法中用到，為了防止通知的URL變化結(jié)果為空。
4.doSaveProperties
該方法比較長(zhǎng)，我這里不貼源碼了，需要的就查看github上的分析，該方法主要是將內(nèi)存緩存properties中的數(shù)據(jù)存儲(chǔ)到文件中，并且在里面做了版本號(hào)的控制，防止老的版本數(shù)據(jù)覆蓋了新版本數(shù)據(jù)。數(shù)據(jù)流向是跟loadProperties方法相反。
5.loadProperties
private void loadProperties() {
if (file != null && file.exists()) {
InputStream in = null;
try {
in = new FileInputStream(file);
// 把數(shù)據(jù)寫(xiě)入到內(nèi)存緩存中
properties.load(in);
if (logger.isInfoEnabled()) {
logger.info("Load registry store file " + file + ", data: " + properties);
}
} catch (Throwable e) {
logger.warn("Failed to load registry store file " + file, e);
} finally {
if (in != null) {
try {
in.close();
} catch (IOException e) {
logger.warn(e.getMessage(), e);
}
}
}
}
}
該方法就是加載本地磁盤(pán)緩存文件到內(nèi)存緩存，也就是把文件里面的數(shù)據(jù)寫(xiě)入properties，可以對(duì)比doSaveProperties方法，其中關(guān)鍵的實(shí)現(xiàn)就是properties.load和properties.store的區(qū)別，邏輯并不難。跟doSaveProperties的數(shù)據(jù)流向相反。
6.getCacheUrls
public List getCacheUrls(URL url) {
for (Map.Entry entry : properties.entrySet()) {
// key為某個(gè)分類(lèi)，例如服務(wù)提供者分類(lèi)
String key = (String) entry.getKey();
// value為某個(gè)分類(lèi)的列表，例如服務(wù)提供者列表
String value = (String) entry.getValue();
if (key != null && key.length() > 0 && key.equals(url.getServiceKey())
&& (Character.isLetter(key.charAt(0)) || key.charAt(0) == "_")
&& value != null && value.length() > 0) {
//分割出列表的每個(gè)值
String[] arr = value.trim().split(URL_SPLIT);
List urls = new ArrayList();
for (String u : arr) {
urls.add(URL.valueOf(u));
}
return urls;
}
}
return null;
}
該方法是獲得內(nèi)存緩存properties中相關(guān)value，并且返回為一個(gè)集合，從該方法中可以很清楚的看出properties中是存儲(chǔ)的什么數(shù)據(jù)格式。
7.lookup
來(lái)看看源碼：
@Override
public List lookup(URL url) {
List result = new ArrayList();
// 獲得該消費(fèi)者url訂閱的 所有被通知的 服務(wù)URL集合
Map> notifiedUrls = getNotified().get(url);
// 判斷該消費(fèi)者是否訂閱服務(wù)
if (notifiedUrls != null && notifiedUrls.size() > 0) {
for (List urls : notifiedUrls.values()) {
for (URL u : urls) {
// 判斷協(xié)議是否為空
if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
// 添加 該消費(fèi)者訂閱的服務(wù)URL
result.add(u);
}
}
}
} else {
// 原子類(lèi) 避免在獲取注冊(cè)在注冊(cè)中心的服務(wù)url時(shí)能夠保證是最新的url集合
final AtomicReference> reference = new AtomicReference>();
// 通知監(jiān)聽(tīng)器。當(dāng)收到服務(wù)變更通知時(shí)觸發(fā)
NotifyListener listener = new NotifyListener() {
@Override
public void notify(List urls) {
reference.set(urls);
}
};
// 訂閱服務(wù)，就是消費(fèi)者url訂閱已經(jīng) 注冊(cè)在注冊(cè)中心的服務(wù)（也就是添加該服務(wù)的監(jiān)聽(tīng)器）
subscribe(url, listener); // Subscribe logic guarantees the first notify to return
List urls = reference.get();
if (urls != null && !urls.isEmpty()) {
for (URL u : urls) {
if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
result.add(u);
}
}
}
}
return result;
}
該方法是實(shí)現(xiàn)了RegistryService接口的方法，作用是獲得消費(fèi)者url訂閱的服務(wù)URL列表。該方法有幾個(gè)地方有些繞我在這里重點(diǎn)講解一下：
URL可能是消費(fèi)者URL，也可能是注冊(cè)在注冊(cè)中心的服務(wù)URL，我在注釋中在URL加了修飾，為了能更明白的區(qū)分。
訂閱了的服務(wù)URL一定是在注冊(cè)中心中注冊(cè)了的。
關(guān)于訂閱服務(wù)subscribe方法和通知監(jiān)聽(tīng)器NotifyListener，我會(huì)在下面解釋。
8.register && unregister
這兩個(gè)方法實(shí)現(xiàn)了RegistryService接口的方法，里面的邏輯很簡(jiǎn)單，所有我就不貼代碼了，以免影響篇幅，如果真想看，可以進(jìn)到我github查看，下面我會(huì)貼出這部分注釋github的地址。其中注冊(cè)的邏輯就是把url加入到屬性registered，而取消注冊(cè)的邏輯就是把url從該屬性中移除，該屬性在上面有介紹。真正的實(shí)現(xiàn)是在FailbackRegistry類(lèi)中，F(xiàn)ailbackRegistry類(lèi)我會(huì)在下面介紹。
9.subscribe && unsubscribe
這兩個(gè)方法實(shí)現(xiàn)了RegistryService接口的方法，分別是訂閱和取消訂閱，我就貼一個(gè)訂閱的代碼：
@Override
public void subscribe(URL url, NotifyListener listener) {
if (url == null) {
throw new IllegalArgumentException("subscribe url == null");
}
if (listener == null) {
throw new IllegalArgumentException("subscribe listener == null");
}
if (logger.isInfoEnabled()) {
logger.info("Subscribe: " + url);
}
// 獲得該消費(fèi)者url 已經(jīng)訂閱的服務(wù) 的監(jiān)聽(tīng)器集合
Set listeners = subscribed.get(url);
if (listeners == null) {
subscribed.putIfAbsent(url, new ConcurrentHashSet());
listeners = subscribed.get(url);
}
// 添加某個(gè)服務(wù)的監(jiān)聽(tīng)器
listeners.add(listener);
}
從源代碼可以看到，其實(shí)訂閱也就是把服務(wù)通知監(jiān)聽(tīng)器加入到subscribed中，具體的實(shí)現(xiàn)也是在FailbackRegistry類(lèi)中。
10.recover
恢復(fù)方法，在注冊(cè)中心斷開(kāi)，重連成功的時(shí)候，會(huì)恢復(fù)注冊(cè)和訂閱。
protected void recover() throws Exception {
// register
//把內(nèi)存緩存中的registered取出來(lái)遍歷進(jìn)行注冊(cè)
Set recoverRegistered = new HashSet(getRegistered());
if (!recoverRegistered.isEmpty()) {
if (logger.isInfoEnabled()) {
logger.info("Recover register url " + recoverRegistered);
}
for (URL url : recoverRegistered) {
register(url);
}
}
// subscribe
//把內(nèi)存緩存中的subscribed取出來(lái)遍歷進(jìn)行訂閱
Map> recoverSubscribed = new HashMap>(getSubscribed());
if (!recoverSubscribed.isEmpty()) {
if (logger.isInfoEnabled()) {
logger.info("Recover subscribe url " + recoverSubscribed.keySet());
}
for (Map.Entry> entry : recoverSubscribed.entrySet()) {
URL url = entry.getKey();
for (NotifyListener listener : entry.getValue()) {
subscribe(url, listener);
}
}
}
}
11.notify
protected void notify(List urls) {
if (urls == null || urls.isEmpty()) return;
// 遍歷訂閱URL的監(jiān)聽(tīng)器集合，通知他們
for (Map.Entry> entry : getSubscribed().entrySet()) {
URL url = entry.getKey();
// 匹配
if (!UrlUtils.isMatch(url, urls.get(0))) {
continue;
}
// 遍歷監(jiān)聽(tīng)器集合，通知他們
Set listeners = entry.getValue();
if (listeners != null) {
for (NotifyListener listener : listeners) {
try {
notify(url, listener, filterEmpty(url, urls));
} catch (Throwable t) {
logger.error("Failed to notify registry event, urls: " + urls + ", cause: " + t.getMessage(), t);
}
}
}
}
}
protected void notify(URL url, NotifyListener listener, List urls) {
if (url == null) {
throw new IllegalArgumentException("notify url == null");
}
if (listener == null) {
throw new IllegalArgumentException("notify listener == null");
}
if ((urls == null || urls.isEmpty())
&& !Constants.ANY_VALUE.equals(url.getServiceInterface())) {
logger.warn("Ignore empty notify urls for subscribe url " + url);
return;
}
if (logger.isInfoEnabled()) {
logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
}
Map> result = new HashMap>();
// 將urls進(jìn)行分類(lèi)
for (URL u : urls) {
if (UrlUtils.isMatch(url, u)) {
// 按照url中key為category對(duì)應(yīng)的值進(jìn)行分類(lèi)，如果沒(méi)有該值，就找key為providers的值進(jìn)行分類(lèi)
String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
List categoryList = result.get(category);
if (categoryList == null) {
categoryList = new ArrayList();
// 分類(lèi)結(jié)果放入result
result.put(category, categoryList);
}
categoryList.add(u);
}
}
if (result.size() == 0) {
return;
}
// 獲得某一個(gè)消費(fèi)者被通知的url集合（通知的 URL 變化結(jié)果）
Map> categoryNotified = notified.get(url);
if (categoryNotified == null) {
// 添加該消費(fèi)者對(duì)應(yīng)的url
notified.putIfAbsent(url, new ConcurrentHashMap>());
categoryNotified = notified.get(url);
}
// 處理通知監(jiān)聽(tīng)器URL 變化結(jié)果
for (Map.Entry> entry : result.entrySet()) {
String category = entry.getKey();
List categoryList = entry.getValue();
// 把分類(lèi)標(biāo)實(shí)和分類(lèi)后的列表放入notified的value中
// 覆蓋到 `notified`
// 當(dāng)某個(gè)分類(lèi)的數(shù)據(jù)為空時(shí)，會(huì)依然有 urls 。其中 `urls[0].protocol = empty` ，通過(guò)這樣的方式，處理所有服務(wù)提供者為空的情況。
categoryNotified.put(category, categoryList);
// 保存到文件
saveProperties(url);
//通知監(jiān)聽(tīng)器
listener.notify(categoryList);
}
}
notify方法是通知監(jiān)聽(tīng)器，url的變化結(jié)果，不過(guò)變化的是全量數(shù)據(jù)，全量數(shù)據(jù)意思就是是以服務(wù)接口和數(shù)據(jù)類(lèi)型為維度全量通知，即不會(huì)通知一個(gè)服務(wù)的同類(lèi)型的部分?jǐn)?shù)據(jù)，用戶(hù)不需要對(duì)比上一次通知結(jié)果。這里要注意幾個(gè)重點(diǎn)：
發(fā)起訂閱后，會(huì)獲取全量數(shù)據(jù)，此時(shí)會(huì)調(diào)用notify方法。即Registry 獲取到了全量數(shù)據(jù)
每次注冊(cè)中心發(fā)生變更時(shí)會(huì)調(diào)用notify方法雖然變化是增量，調(diào)用這個(gè)方法的調(diào)用方，已經(jīng)進(jìn)行處理，傳入的urls依然是全量的。
listener.notify，通知監(jiān)聽(tīng)器，例如，有新的服務(wù)提供者啟動(dòng)時(shí)，被通知，創(chuàng)建新的 Invoker 對(duì)象。
12.saveProperties
先來(lái)看看源碼：
private void saveProperties(URL url) {
if (file == null) {
return;
}
try {
// 拼接url
StringBuilder buf = new StringBuilder();
Map> categoryNotified = notified.get(url);
if (categoryNotified != null) {
for (List us : categoryNotified.values()) {
for (URL u : us) {
if (buf.length() > 0) {
buf.append(URL_SEPARATOR);
}
buf.append(u.toFullString());
}
}
}
// 設(shè)置到properties中
properties.setProperty(url.getServiceKey(), buf.toString());
// 增加版本號(hào)
long version = lastCacheChanged.incrementAndGet();
if (syncSaveFile) {
// 將集合中的數(shù)據(jù)存儲(chǔ)到文件中
doSaveProperties(version);
} else {
//異步開(kāi)啟保存到文件
registryCacheExecutor.execute(new SaveProperties(version));
}
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
該方法是單個(gè)消費(fèi)者url對(duì)應(yīng)在notified中的數(shù)據(jù)，保存在到文件，而保存到文件的操作是調(diào)用了doSaveProperties方法，該方法跟doSaveProperties的區(qū)別是doSaveProperties方法將properties數(shù)據(jù)全部覆蓋性的保存到文件，而saveProperties只是保存單個(gè)消費(fèi)者url的數(shù)據(jù)。
13.destroy
該方法在JVM關(guān)閉時(shí)調(diào)用，進(jìn)行取消注冊(cè)和訂閱的操作。具體邏輯就是調(diào)用了unregister和unsubscribe方法，有需要看源碼的可以進(jìn)入github查看。
（六）support包下的FailbackRegistry
我在上面講AbstractRegistry類(lèi)的時(shí)候已經(jīng)提到了FailbackRegistry，F(xiàn)ailbackRegistry繼承了AbstractRegistry，AbstractRegistry中的注冊(cè)訂閱等方法，實(shí)際上就是一些內(nèi)存緩存的變化，而真正的注冊(cè)訂閱的實(shí)現(xiàn)邏輯在FailbackRegistry實(shí)現(xiàn)，并且FailbackRegistry提供了失敗重試的機(jī)制。
源碼注釋地址：https://github.com/CrazyHZM/i...
1.屬性
// Scheduled executor service
// 定時(shí)任務(wù)執(zhí)行器
private final ScheduledExecutorService retryExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryFailedRetryTimer", true));
// Timer for failure retry, regular check if there is a request for failure, and if there is, an unlimited retry
// 失敗重試定時(shí)器，定時(shí)去檢查是否有請(qǐng)求失敗的，如有，無(wú)限次重試。
private final ScheduledFuture retryFuture;
// 注冊(cè)失敗的URL集合
private final Set failedRegistered = new ConcurrentHashSet();
// 取消注冊(cè)失敗的URL集合
private final Set failedUnregistered = new ConcurrentHashSet();
// 訂閱失敗的監(jiān)聽(tīng)器集合
private final ConcurrentMap> failedSubscribed = new ConcurrentHashMap>();
// 取消訂閱失敗的監(jiān)聽(tīng)器集合
private final ConcurrentMap> failedUnsubscribed = new ConcurrentHashMap>();
// 通知失敗的URL集合
private final ConcurrentMap>> failedNotified = new ConcurrentHashMap>>();
該類(lèi)的屬性比較好理解，也可以很明顯看出這些屬性都是跟失敗重試機(jī)制相關(guān)。
2.構(gòu)造函數(shù)
public FailbackRegistry(URL url) {
super(url);
// 從url中讀取重試頻率，如果為空，則默認(rèn)5000ms
this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
// 創(chuàng)建失敗重試定時(shí)器
this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
// Check and connect to the registry
try {
//重試
retry();
} catch (Throwable t) { // Defensive fault tolerance
logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
}
}
}, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
}
構(gòu)造函數(shù)主要是創(chuàng)建了失敗重試的定時(shí)器，重試頻率從URL取，如果沒(méi)有設(shè)置，則默認(rèn)為5000ms。
3.register && unregister && subscribe && unsubscribe
這四個(gè)方法就是注冊(cè)、取消注冊(cè)、訂閱、取消訂閱的具體實(shí)現(xiàn)，因?yàn)榇a邏輯極其相似，所以為放在一起，下面為只貼出注冊(cè)的源碼：
public void register(URL url) {
super.register(url);
//首先從失敗的緩存中刪除該url
failedRegistered.remove(url);
failedUnregistered.remove(url);
try {
// Sending a registration request to the server side
// 向注冊(cè)中心發(fā)送一個(gè)注冊(cè)請(qǐng)求
doRegister(url);
} catch (Exception e) {
Throwable t = e;
// If the startup detection is opened, the Exception is thrown directly.
// 如果開(kāi)啟了啟動(dòng)時(shí)檢測(cè)，則直接拋出異常
boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
&& url.getParameter(Constants.CHECK_KEY, true)
&& !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
boolean skipFailback = t instanceof SkipFailbackWrapperException;
if (check || skipFailback) {
if (skipFailback) {
t = t.getCause();
}
throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
} else {
logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
// Record a failed registration request to a failed list, retry regularly
// 把這個(gè)注冊(cè)失敗的url放入緩存，并且定時(shí)重試。
failedRegistered.add(url);
}
}
可以看到，邏輯很清晰，就是做了一個(gè)doRegister的操作，如果失敗拋出異常，則加入到失敗的緩存中進(jìn)行重試。為這里要解釋的是doRegister，與之對(duì)應(yīng)的還有doUnregister、doSubscribe、doUnsubscribe三個(gè)方法，是FailbackRegistry抽象出來(lái)的方法，意圖在于每種實(shí)現(xiàn)注冊(cè)中心的方法不一樣，相對(duì)應(yīng)的注冊(cè)、訂閱等操作也會(huì)有所區(qū)別，而把這四個(gè)方法抽象出現(xiàn)，為了讓子類(lèi)只去關(guān)注這四個(gè)的實(shí)現(xiàn)，比如說(shuō)redis實(shí)現(xiàn)的注冊(cè)中心跟zookeeper實(shí)現(xiàn)的注冊(cè)中心方式肯定不一樣，那么對(duì)應(yīng)的注冊(cè)訂閱等操作也有所不同，那么各自只要去實(shí)現(xiàn)該抽象方法即可。
其他的三個(gè)方法有需要的可以查看github上的我寫(xiě)的注釋。
4.notify
@Override
protected void notify(URL url, NotifyListener listener, List urls) {
if (url == null) {
throw new IllegalArgumentException("notify url == null");
}
if (listener == null) {
throw new IllegalArgumentException("notify listener == null");
}
try {
// 通知 url 數(shù)據(jù)變化
doNotify(url, listener, urls);
} catch (Exception t) {
// Record a failed registration request to a failed list, retry regularly
// 放入失敗的緩存中，重試
Map> listeners = failedNotified.get(url);
if (listeners == null) {
failedNotified.putIfAbsent(url, new ConcurrentHashMap>());
listeners = failedNotified.get(url);
}
listeners.put(listener, urls);
logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
}
protected void doNotify(URL url, NotifyListener listener, List urls) {
super.notify(url, listener, urls);
}
可以看到notify不一樣，他還是又回去調(diào)用了父類(lèi)AbstractRegistry的notify，與上述四個(gè)方法不一樣。
5.revocer
@Override
protected void recover() throws Exception {
// register
// register 恢復(fù)注冊(cè)，添加到 `failedRegistered` ，定時(shí)重試
Set recoverRegistered = new HashSet(getRegistered());
if (!recoverRegistered.isEmpty()) {
if (logger.isInfoEnabled()) {
logger.info("Recover register url " + recoverRegistered);
}
for (URL url : recoverRegistered) {
failedRegistered.add(url);
}
}
// subscribe
// subscribe 恢復(fù)訂閱，添加到 `failedSubscribed` ，定時(shí)重試
Map> recoverSubscribed = new HashMap>(getSubscribed());
if (!recoverSubscribed.isEmpty()) {
if (logger.isInfoEnabled()) {
logger.info("Recover subscribe url " + recoverSubscribed.keySet());
}
for (Map.Entry> entry : recoverSubscribed.entrySet()) {
URL url = entry.getKey();
for (NotifyListener listener : entry.getValue()) {
addFailedSubscribed(url, listener);
}
}
}
}
重寫(xiě)了父類(lèi)的recover，將注冊(cè)和訂閱放入到對(duì)應(yīng)的失敗緩存中，然后定時(shí)重試。
6.retry
該方法中實(shí)現(xiàn)了重試的邏輯，分別對(duì)注冊(cè)失敗failedRegistered、取消注冊(cè)失敗failedUnregistered、訂閱失敗failedSubscribed、取消訂閱失敗failedUnsubscribed、通知監(jiān)聽(tīng)器失敗failedNotified這五個(gè)緩存中的元素進(jìn)行重試，重試的邏輯就是調(diào)用了相關(guān)的方法，然后從緩存中刪除，例如重試注冊(cè)，先進(jìn)行doRegister，然后把該url從failedRegistered移除。具體的注釋請(qǐng)到GitHub查看。
（七）support包下的AbstractRegistryFactory
該類(lèi)實(shí)現(xiàn)了RegistryFactory接口，抽象了createRegistry方法，它實(shí)現(xiàn)了Registry的容器管理。
1.屬性
// Log output
// 日志記錄
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractRegistryFactory.class);
// The lock for the acquisition process of the registry
// 鎖，對(duì)REGISTRIES訪問(wèn)對(duì)競(jìng)爭(zhēng)控制
private static final ReentrantLock LOCK = new ReentrantLock();
// Registry Collection Map
// Registry 集合
private static final Map REGISTRIES = new ConcurrentHashMap();
2.destroyAll
public static void destroyAll() {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Close all registries " + getRegistries());
}
// Lock up the registry shutdown process
// 獲得鎖
LOCK.lock();
try {
for (Registry registry : getRegistries()) {
try {
// 銷(xiāo)毀
registry.destroy();
} catch (Throwable e) {
LOGGER.error(e.getMessage(), e);
}
}
// 清空緩存
REGISTRIES.clear();
} finally {
// Release the lock
// 釋放鎖
LOCK.unlock();
}
}
該方法作用是銷(xiāo)毀所有的Registry對(duì)象，并且清除內(nèi)存緩存，邏輯比較簡(jiǎn)單，關(guān)鍵就是對(duì)REGISTRIES進(jìn)行同步的操作。
3.getRegistry
@Override
public Registry getRegistry(URL url) {
// 修改url
url = url.setPath(RegistryService.class.getName())
.addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
.removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
// 計(jì)算key值
String key = url.toServiceString();
// Lock the registry access process to ensure a single instance of the registry
// 獲得鎖
LOCK.lock();
try {
Registry registry = REGISTRIES.get(key);
if (registry != null) {
return registry;
}
// 創(chuàng)建Registry對(duì)象
registry = createRegistry(url);
if (registry == null) {
throw new IllegalStateException("Can not create registry " + url);
}
// 添加到緩存。
REGISTRIES.put(key, registry);
return registry;
} finally {
// Release the lock
// 釋放鎖
LOCK.unlock();
}
}
該方法是實(shí)現(xiàn)了RegistryFactory接口中的方法，關(guān)于key值的計(jì)算我會(huì)在后續(xù)講解URL的文章中講到，這里最要注意的是createRegistry，因?yàn)锳bstractRegistryFactory類(lèi)把這個(gè)方法抽象出來(lái)，為了讓子類(lèi)只要關(guān)注該方法，比如說(shuō)redis實(shí)現(xiàn)的注冊(cè)中心和zookeeper實(shí)現(xiàn)的注冊(cè)中心創(chuàng)建方式肯定不同，而他們相同的一些操作都已經(jīng)在AbstractRegistryFactory中實(shí)現(xiàn)。所以只要關(guān)注并且實(shí)現(xiàn)該抽象方法即可。
（八）support包下的ConsumerInvokerWrapper && ProviderInvokerWrapper
這兩個(gè)類(lèi)實(shí)現(xiàn)了Invoker接口，分別是服務(wù)消費(fèi)者和服務(wù)提供者的Invoker的包裝器，其中就包裝了一些屬性，我們來(lái)看看源碼：
1.ConsumerInvokerWrapper屬性
// Invoker 對(duì)象
private Invoker invoker;
// 原始url
private URL originUrl;
// 注冊(cè)中心url
private URL registryUrl;
// 消費(fèi)者url
private URL consumerUrl;
// 注冊(cè)中心 Directory
private RegistryDirectory registryDirectory;
2.ProviderInvokerWrapper屬性
// Invoker對(duì)象
private Invoker invoker;
// 原始url
private URL originUrl;
// 注冊(cè)中心url
private URL registryUrl;
// 服務(wù)提供者url
private URL providerUrl;
// 是否注冊(cè)
private volatile boolean isReg;
這兩個(gè)類(lèi)都被運(yùn)用在Dubbo QOS中，需要了解Dubbo QOS的可以到官方文檔里面查看
QOS網(wǎng)址：http://dubbo.apache.org/zh-cn...
（九）support包下的ProviderConsumerRegTable
服務(wù)提供者和消費(fèi)者注冊(cè)表，存儲(chǔ)JVM進(jìn)程中服務(wù)提供者和消費(fèi)者的Invoker，該類(lèi)也是被運(yùn)用在QOS中，包括上面的兩個(gè)類(lèi)，都跟QOS中的Offline下線服務(wù)命令和ls列出消費(fèi)者和提供者邏輯實(shí)現(xiàn)有關(guān)系。我們可以看看它的屬性：
// 服務(wù)提供者Invoker集合，key 為服務(wù)提供者的url 計(jì)算的key，就是url.toServiceString()方法得到的
public static ConcurrentHashMap> providerInvokers = new ConcurrentHashMap>();
// 服務(wù)消費(fèi)者的Invoker集合，key 為服務(wù)消費(fèi)者的url 計(jì)算的key，url.toServiceString()方法得到的
public static ConcurrentHashMap> consumerInvokers = new ConcurrentHashMap>();
可以看到，其實(shí)記錄的服務(wù)提供者、消費(fèi)者、注冊(cè)中心中間的調(diào)用鏈，為了從一方出發(fā)能夠很直觀的找到跟它相關(guān)聯(lián)的所有調(diào)用鏈。
該類(lèi)中的其他方法請(qǐng)自行查看，這部分跟運(yùn)維命令的實(shí)現(xiàn)相關(guān)，所以為不在這里講解。
（十）support包下的SkipFailbackWrapperException
該類(lèi)是一個(gè)dubbo多帶帶創(chuàng)建的異常，在FailbackRegistry中被使用到，自定義的是一個(gè)跳過(guò)失敗重試的異常。
（十一）status包下的RegistryStatusChecker
該類(lèi)實(shí)現(xiàn)了StatusChecker，StatusChecker是一個(gè)狀態(tài)校驗(yàn)的接口，RegistryStatusChecker是它的擴(kuò)展類(lèi)，做了一些跟注冊(cè)中心有關(guān)的狀態(tài)檢查和設(shè)置。我們來(lái)看看源碼：
@Activate
public class RegistryStatusChecker implements StatusChecker {
@Override
public Status check() {
// 獲得所有的注冊(cè)中心對(duì)象
Collection registries = AbstractRegistryFactory.getRegistries();
if (registries.isEmpty()) {
return new Status(Status.Level.UNKNOWN);
}
Status.Level level = Status.Level.OK;
StringBuilder buf = new StringBuilder();
// 拼接注冊(cè)中心url中的地址
for (Registry registry : registries) {
if (buf.length() > 0) {
buf.append(",");
}
buf.append(registry.getUrl().getAddress());
// 如果注冊(cè)中心的節(jié)點(diǎn)不可用，則拼接disconnected，并且狀態(tài)設(shè)置為error
if (!registry.isAvailable()) {
level = Status.Level.ERROR;
buf.append("(disconnected)");
} else {
buf.append("(connected)");
}
}
// 返回狀態(tài)檢查結(jié)果
return new Status(level, buf.toString());
}
}
第一個(gè)關(guān)注點(diǎn)就是@Activate注解，也就是RegistryStatusChecker類(lèi)會(huì)自動(dòng)激活加載。該類(lèi)就實(shí)現(xiàn)了接口的check方法，作用就是給注冊(cè)中心進(jìn)行狀態(tài)檢查，并且返回檢查結(jié)果。
下面講的是integration下面的兩個(gè)類(lèi)RegistryProtocol和RegistryDirectory，這兩個(gè)類(lèi)與注冊(cè)中心核心的邏輯關(guān)系沒(méi)有那么強(qiáng)。RegistryProtocol是對(duì)dubbo-rpc-api的依賴(lài)集成，RegistryDirectory是對(duì)dubbo-cluster的依賴(lài)集成。如果看了下面的解析有點(diǎn)糊涂，可以先跳過(guò)這部分，等我出了rpc和cluster相關(guān)的文章后再回來(lái)看就會(huì)比較清晰。
（十二）integration包下的RegistryProtocol && RegistryDirectory
RegistryProtocol實(shí)現(xiàn)了Protocol接口，也是Protocol接口等擴(kuò)展類(lèi)，但是它可以認(rèn)為并不是一個(gè)真正的協(xié)議，他是實(shí)際的協(xié)議（dubbo . rmi）包裝者，這樣客戶(hù)端的請(qǐng)求在一開(kāi)始如果沒(méi)有服務(wù)端的信息，會(huì)先從注冊(cè)中心拉取服務(wù)的注冊(cè)信息，然后再和服務(wù)端直連。RegistryProtocol是基于注冊(cè)中心發(fā)現(xiàn)服務(wù)提供者的實(shí)現(xiàn)協(xié)議。
RegistryDirectory：注冊(cè)中心服務(wù)，維護(hù)著所有可用的遠(yuǎn)程Invoker或者本地的Invoker。 它的Invoker集合是從注冊(cè)中心獲取的， 它實(shí)現(xiàn)了NotifyListener接口實(shí)現(xiàn)了回調(diào)接口notify方法。比如消費(fèi)方要調(diào)用某遠(yuǎn)程服務(wù)，會(huì)向注冊(cè)中心訂閱這個(gè)服務(wù)的所有服務(wù)提供方，訂閱時(shí)和服務(wù)提供方數(shù)據(jù)有變動(dòng)時(shí)回調(diào)消費(fèi)方的NotifyListener服務(wù)的notify方法，回調(diào)接口傳入所有服務(wù)的提供方的url地址然后將urls轉(zhuǎn)化為invokers, 也就是refer應(yīng)用遠(yuǎn)程服務(wù)。
這兩個(gè)類(lèi)等我講解完rpc和cluster模塊之后再進(jìn)行補(bǔ)充源碼解析。
后記
該部分相關(guān)的源碼解析地址：https://github.com/CrazyHZM/i...
該文章講解了dubbo的注冊(cè)中心關(guān)于服務(wù)注冊(cè)、訂閱、服務(wù)變更通知等內(nèi)部邏輯實(shí)現(xiàn)，接下來(lái)四篇文章我將會(huì)講解dubbo、multicast、zookeeper、redis四種實(shí)現(xiàn)注冊(cè)中心策略的邏輯實(shí)現(xiàn)。如果我在哪一部分寫(xiě)的不夠到位或者寫(xiě)錯(cuò)了，歡迎給我提意見(jiàn)，我的私人微信號(hào)碼：HUA799695226。
