Dubbo 微服務系列（03）服務注冊

目錄

• Dubbo 微服務系列（03）服務注冊
  • Spring Cloud Alibaba 系列目錄 - Dubbo 篇
  • 1. 背景介紹
  • 2. 數據結構
  • 3. 源碼分析
    • 3.1 緩存機制
      • 3.1.1 緩存的加載
      • 3.1.2 緩存的更新
    • 3.2 重試機制
    • 3.3 ZookeeperRegistry
      • 3.3.1 初始化
      • 3.3.2 注冊
      • 3.3.3 訂閱
    • 3.4 NacosRegistry
      • 3.4.1 注冊
      • 3.4.2 訂閱
    • 4. 總結
    • 4.1 服務自省

Spring Cloud Alibaba 系列目錄 - Dubbo 篇

1. 背景介紹

圖1 Dubbo經典架構圖

注：本圖來源 Dubbo官方架構圖

表1 節點角色說明

節點	角色說明
Provider	暴露服務的服務提供方
Consumer	調用遠程服務的服務消費方
Registry	服務注冊與發現的注冊中心
Monitor	統計服務的調用次數和調用時間的監控中心
Container	服務運行容器

在 Dubbo 微服務體系中，注冊中心是其核心組件之一，Dubbo 通過注冊中心實現服務的注冊與發現。Dubbo 的注冊中心有 zookeeper、nacos 等。

• dubbo-registry-api 注冊中心抽象 API
• dubbo-registry-default Dubbo 基於內存的默認實現
• dubbo-registry-multicast multicast 注冊中心
• dubbo-registry-zookeeper zookeeper 注冊中心
• dubbo-registry-nacos nacos 注冊中心

2. 數據結構

不同的注冊中心，數據結構稍有區別。下面以 Zookeeper 為例，說明 dubbo 注冊中心的數據結構。

（1）路徑結構

例如：/dubbo/com.dubbo.DemoService1/providers 是服務都在 ZK 上的注冊路徑，該路徑結構分為四層：

• 一是root（根節點，默認為 /dubbo）；
• 二是 serviceInterface（接口名稱）；
• 三是服務類型（providers、consumers、routers、configurators）；
• 四是具體注冊的元信息 URL。

注： 前三層相當於 serviceKey，最后一層則是對應的 serviceValue。

+ /dubbo
+-- serviceInterface
	+-- providers
	+-- consumers
	+-- routers
	+-- configurators

（2）四種服務類型（category）分別是 providers、consumers、routers、configurators：

• /dubbo/serviceInterface/providers 服務提供者注冊信息，包含多個服務者 URL 元數據信息。eg: dubbo://192.168.139.101:20880/com.dubbo.DemoService1?key=value&...
• /dubbo/serviceInterface/consumers 服務消費才注冊信息，包含多個消費者 URL 元數據信息。eg: dubbo://192.168.139.101:8888/com.dubbo.DemoService1?key=value&...
• /dubbo/serviceInterface/router 路由配置信息，包含消費者路由策略 URL 元數據信息。eg: condition://0.0.0.0/com.dubbo.DemoService1?category=routers&key=value&...
• /dubbo/serviceInterface/configurators 外部化配置信息，包含服務者動態配置 URL 元數據信息。eg: override://0.0.0.0/com.dubbo.DemoService1?category=configurators&key=value&...

圖2 Zookeeper 注冊中心數據結構

graph TB ROOT((/dubbo)) --> DemoService1(com.deme.DemoService1) ROOT --> DemoService2(com.deme.DemoService2) DemoService1 -.-> providers(providers) DemoService1 -.-> consumers(consumers) providers -.-> dubbo://192.168.139.101:20080 DemoService2 -.-> routes(routes) DemoService2 -.-> configurators(configurators) configurators -.-> override://0.0.0.0/...

3. 源碼分析

圖3 Dubbo注冊中心類圖

• AbstractRegistry 緩存機制
• FailbackRegistry 重試機制
• ZookeeperRegistry、NacosRegistry 具體的注冊中心實現，每個注冊中心都有一個對應的工廠類，如 ZookeeperRegistryFactory、NacosRegistryFactory。當消費者 URL 訂閱的注冊信息發生變化時，ZookeeperRegistry 會回調 notify(URL url, NotifyListener listener, List<URL> urls) 方法，更新內存和磁盤上本地緩存的注冊信息，並通知監聽者。

public interface RegistryService {
	
    // 注冊服務
    // dubbo://10.20.153.10/dubbo.BarService?version=1.0.0&application=kylin
    void register(URL url);
    void unregister(URL url);

    // 訂閱指定服務
    // consumer://10.20.153.10/dubbo.BarService?version=1.0.0&application=kylin
    void subscribe(URL url, NotifyListener listener);
    void unsubscribe(URL url, NotifyListener listener);

    // 查找指定服務
    // consumer://10.20.153.10/dubbo.BarService?version=1.0.0&application=kylin
    List<URL> lookup(URL url);
}

3.1 緩存機制

消費者從注冊中心獲取注冊信息后會做本地緩存，本地緩存保存兩份：一是內存中保存一份，通過 notified 的 Map 結構進行保存；二是磁盤上保存一份，通過 file 保持引用。

private final Properties properties = new Properties();
private File file;				// 磁盤文件服務緩存對象
private final ConcurrentMap<URL, Map<String, List<URL>>> notified = 
    new ConcurrentHashMap<>();	// 內存中的服務緩存對象

• notified 是內存中的服務緩存對象，外層 Key 是消費者 URL，內層的 kye 是分類（category），包含 providers、consumers、routers、configurators 四種，value 則是對應的服務列表。
• file 磁盤緩存對象，當訂閱的信息發生變更時先更新 properties 的內容，通過 properties 再寫入磁盤。

3.1.1 緩存的加載

當初始化注冊中心時，會通過 AbstractRegistry 的默認構造器加載磁盤緩存文件 file 中的訂閱信息。當注冊中心無法連接或宕機時使用緩存。

// 初始化加載磁盤緩存文件 file 中的訂閱信息
private void loadProperties() {
    if (file != null && file.exists()) {
        ...
        InputStream in = new FileInputStream(file);
        properties.load(in);
    }
}

3.1.2 緩存的更新

當訂閱的注冊信息發生變量時，ZookeeperRegistry 會回調 notify 方法更新緩存中的數據，其中第一個參數為消費者 url，第三個參數為注冊中心注冊的 urls。

/**
 * 當訂閱的注冊信息發生變更時，通知 consumer url 更新注冊列表
 *
 * @param url      consumer side url
 * @param listener listener
 * @param urls     provider latest urls
 */
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    ...
    // 按category進行分類，並根據消費者url過濾訂閱的urls
    Map<String, List<URL>> result = new HashMap<>();
    for (URL u : urls) {
        if (UrlUtils.isMatch(url, u)) {
            String category = u.getParameter(CATEGORY_KEY, DEFAULT_CATEGORY);
            List<URL> categoryList = result.computeIfAbsent(category, k -> new ArrayList<>());
            categoryList.add(u);
        }
    }
    if (result.size() == 0) {
        return;
    }
    Map<String, List<URL>> categoryNotified =
        notified.computeIfAbsent(url, u -> new ConcurrentHashMap<>());
    for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
        String category = entry.getKey();
        List<URL> categoryList = entry.getValue();
        // 1. 更新內存中的本地緩存：notified
        categoryNotified.put(category, categoryList);
        // 2. 更新磁盤中的本地緩存：properties -> file
        saveProperties(url);
        // 3. 通知監聽者
        listener.notify(categoryList);
    }
}

總結： 當注冊信息發生變量時，主要做了三件事：一是更新內存中的注冊信息 notified；二是更新磁盤中的數據 properties；三是通知監聽者。

// 參數url為消費者url，將內存中 notified 對應的消費者 url 對應的注冊信息緩存到磁盤上。
private void saveProperties(URL url) {
    StringBuilder buf = new StringBuilder();
    // 1. 將 notified 對應的 url 注冊信息保存為字符串，用於持久化
    Map<String, List<URL>> categoryNotified = notified.get(url);
    if (categoryNotified != null) {
        for (List<URL> us : categoryNotified.values()) {
            for (URL u : us) {
                if (buf.length() > 0) {
                    buf.append(URL_SEPARATOR);
                }
                buf.append(u.toFullString());
            }
        }
    }
    // 2. 同步到磁盤 file
    properties.setProperty(url.getServiceKey(), buf.toString());
    long version = lastCacheChanged.incrementAndGet();
    if (syncSaveFile) {
        doSaveProperties(version);
    } else {
        registryCacheExecutor.execute(new SaveProperties(version));
    }
}

總結： doSaveProperties 調用 properties.store(outputFile, "Dubbo Registry Cache") 將內存中的注冊信息保存到文件中。properties 的 key、value 分別如下：

• key 消費者 URL#getServiceKey，即 {group/}serviceInterface{:version} ，其中 group 和 version 都是可選。
• value 消費者 URL 訂閱的注冊信息 urls，多個 URL 用空格分隔。示例如下：

"binarylei.dubbo.api.EchoService" -> "empty://192.168.139.1/binarylei.dubbo.api.EchoService?application=dubbo-consumer&category=configurators&dubbo=2.6.0&interface=binarylei.dubbo.api.EchoService&methods=echo&pid=21540&side=consumer&timestamp=1570799586361 empty://192.168.139.1/binarylei.dubbo.api.EchoService?application=dubbo-consumer&category=routers&dubbo=2.6.0&interface=binarylei.dubbo.api.EchoService&methods=echo&pid=21540&side=consumer&timestamp=1570799586361 dubbo://192.168.139.1:20880/binarylei.dubbo.api.EchoService?anyhost=true&application=dubbo-provider&dubbo=2.6.0&generic=false&interface=binarylei.dubbo.api.EchoService&methods=echo&pid=2460&side=provider&timestamp=1570798917842 empty://192.168.139.1/binarylei.dubbo.api.EchoService?application=dubbo-consumer&category=providers,configurators,routers&dubbo=2.6.0&interface=binarylei.dubbo.api.EchoService&methods=echo&pid=21540&side=consumer&timestamp=1570799586361"

3.2 重試機制

FailbackRegistry 繼承自 AbstractRegistry，並在此基礎上增加了失敗重試的能力。FailbackRegistry 內部定義 HashedWheelTimer retryTimer ，會將調用失敗需要重試的任務添加到 retryTimer 中。

// 發起注冊失敗的 URL 集合
private final ConcurrentMap<URL, FailedRegisteredTask> failedRegistered = 
    new ConcurrentHashMap<URL, FailedRegisteredTask>();

// 取消注冊失敗的 URL 集合
private final ConcurrentMap<URL, FailedUnregisteredTask> failedUnregistered =
    new ConcurrentHashMap<URL, FailedUnregisteredTask>();

// 發起訂閱失敗的監聽器集合
private final ConcurrentMap<Holder, FailedSubscribedTask> failedSubscribed =
    new ConcurrentHashMap<Holder, FailedSubscribedTask>();

// 取消訂閱失敗的監聽器集合
private final ConcurrentMap<Holder, FailedUnsubscribedTask> failedUnsubscribed =
    new ConcurrentHashMap<Holder, FailedUnsubscribedTask>();

// 通知失敗的 URL 集合
private final ConcurrentMap<Holder, FailedNotifiedTask> failedNotified =
    new ConcurrentHashMap<Holder, FailedNotifiedTask>();

總結： FailbackRegistry 對注冊、訂閱、通知失敗的情況都進行了重試處理，對於需要重試的任務都保存在對應的集合中，並通過 retryTimer.newTimeout 定時器定時處理。下面以注冊 register 為例分析重試機制。

圖4 Dubbo失敗重試機制

sequenceDiagram participant ZookeeperRegistry participant FailbackRegistry participant FailedRegisteredTask participant AbstractRetryTask participant failedRegistered participant HashedWheelTimer note left of ZookeeperRegistry : register方法 ZookeeperRegistry ->> FailbackRegistry : removeFailedRegistered FailbackRegistry ->> failedRegistered : remove：從failedRegistered集合中刪除重試任務 ZookeeperRegistry ->> FailbackRegistry : removeFailedRegistered FailbackRegistry -->> ZookeeperRegistry : doRegister opt doRegister 注冊失敗 ZookeeperRegistry ->> FailbackRegistry : addFailedRegistered FailbackRegistry ->> FailedRegisteredTask : new FailbackRegistry ->> failedRegistered : putIfAbsent：添加到failedRegistered集合中 FailbackRegistry ->> HashedWheelTimer : newTimeout：如果是新任務，則添加重試任務 opt 重試 HashedWheelTimer -->> AbstractRetryTask : run AbstractRetryTask -->> FailedRegisteredTask : doRetry FailedRegisteredTask -->> FailbackRegistry : doRegister FailedRegisteredTask -->> FailbackRegistry : removeFailedRegisteredTask AbstractRetryTask -->> AbstractRetryTask : reput end end

總結： 當注冊失敗時，Dubbo 會將注冊失敗的 URL 添加到重試任務中。HashedWheelTimer 本質和 Timer 一樣是一個定時器。如果重試成功就會刪除 failedRegistered 隊列中的任務，失敗則調用 reput 繼續重試。在 AbstractRetryTask 配置了兩個默認參數 retryPeriod=5s 和 retryTimes=3，即 5s 重試一次，最多重試 3 次。

@Override
public void register(URL url) {
    super.register(url);
    removeFailedRegistered(url);
    removeFailedUnregistered(url);
    try {
        doRegister(url);
    } catch (Exception e) {
        ...
        // 失敗重試
        addFailedRegistered(url);
    }
}

private void addFailedRegistered(URL url) {
    FailedRegisteredTask oldOne = failedRegistered.get(url);
    if (oldOne != null) {
        return;
    }
    FailedRegisteredTask newTask = new FailedRegisteredTask(url, this);
    oldOne = failedRegistered.putIfAbsent(url, newTask);
    if (oldOne == null) {
        retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
    }
}

3.3 ZookeeperRegistry

ZookeeperRegistry 等具體的實現類，主要功能是實現具體的注冊、訂閱、查找方法 doRegister、doUnregister、doSubscribe、doUnsubscribe、lookup。

3.3.1 初始化

ZookeeperRegistry 初始化主要完成兩件事：一是 zkClient 客戶端初始化；二是注冊監聽器，一旦注冊中心無法連接則將當前注冊和訂閱的 URL 添加到重試任務中。

// url 是注冊中心地址，ZookeeperTransporter 是 ZK 客戶端，默認是 curator
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
    super(url);
    if (url.isAnyHost()) {
        throw new IllegalStateException("registry address == null");
    }
    // 獲取組名，默認為 dubbo
    String group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);
    if (!group.startsWith(PATH_SEPARATOR)) {
        group = PATH_SEPARATOR + group;
    }
    // ZK 注冊的根據路徑是 '/dubbo'
    this.root = group;
    // 創建 Zookeeper 客戶端，默認為 CuratorZookeeperTransporter
    zkClient = zookeeperTransporter.connect(url);
    // 添加狀態監聽器，當 ZK 無法連接時從內存中保存的注冊信息恢復
    zkClient.addStateListener(state -> {
        if (state == StateListener.RECONNECTED) {
            try {
                recover();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    });
}

總結： ZookeeperRegistry 初始化時的兩件事，一是創建客戶端，二是自動恢復。

1. ZookeeperTransporter 客戶端有 curator 和 ZkClient 兩種實現。通過 URL 的 client 或 transporter 進行動態適配，默認的實現是 CuratorZookeeperTransporter。

2. 當注冊中心無法連接時，將當前注冊和訂閱的 URL 添加到重試任務中，一旦網絡正常則自動恢復。recover 是在 FailbackRegistry 中實現的。

@Override
protected void recover() throws Exception {
    // register：將當前注冊的 URL 添加到定時器中進行重試
    Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
    if (!recoverRegistered.isEmpty()) {
        for (URL url : recoverRegistered) {
            addFailedRegistered(url);
        }
    }
    // subscribe：將當前訂閱的 URL 添加到定時器中進行重試
    Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
    if (!recoverSubscribed.isEmpty()) {
        for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
            URL url = entry.getKey();
            for (NotifyListener listener : entry.getValue()) {
                addFailedSubscribed(url, listener);
            }
        }
    }
}

3.3.2 注冊

注冊直接調用 zkClient 的 create 方法創建節點，delete 方法刪除節點，默認為臨時節點。consumer 注冊主要是為了方便 Admin 使用。

@Override
public void doRegister(URL url) {
    try {
        zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
    } catch (Throwable e) {
        throw new RpcException("Failed to register " + url + " to zookeeper " +
                               getUrl() + ", cause: " + e.getMessage(), e);
    }
}

3.3.3 訂閱

訂閱相對注冊復雜很多，分兩種情況，一是 url.getServiceInterface() 是 * ，也就是全量獲取注冊信息，一般是 Admin 使用；二是訂閱指定的 serviceInterface。這里主要分析第二種情況。

@Override
public void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        // 1. 獲取所有的注冊信息，一般 Admin 會獲取所有的服務 ANY_VALUE=*
        if (ANY_VALUE.equals(url.getServiceInterface())) {
            ...
        // 2. 獲取指定的服務 serviceInterface
        } else {
            List<URL> urls = new ArrayList<>();
            // 2.1 '/dubbo/serviceInterface/{providers、routers、configurators}'
            for (String path : toCategoriesPath(url)) {
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                
                // 2.2 創建zkListener，當注冊信息發生變化時，調用notify(url,listener,urls)
                if (zkListener == null) {
                    listeners.putIfAbsent(listener, (parentPath, currentChilds) ->
                    	ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));
                    zkListener = listeners.get(listener);
                }
                // 2.3 創建{providers、routers、configurators}目錄，永久性節點
                zkClient.create(path, false);
                // 2.4 注冊zkListener，並獲取{providers}的子節點信息
                List<String> children = zkClient.addChildListener(path, zkListener);
                if (children != null) {
                    urls.addAll(toUrlsWithEmpty(url, path, children));
                }
            }
            // 2.5 通知 listener
            notify(url, listener, urls);
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " +
                               getUrl() + ", cause: " + e.getMessage(), e);
    }
}

總結： 消費者 URL 會指定需要訂閱的 category{providers、routers、configurators} 類型，依次遍歷這幾個目錄。如果指定的目錄不存在，首先會創建一個永久性的目錄（2.3），並注冊對應的 zkListener（2.4），zkListener 在節點發生變化時調用 notify 通知 listener（2.2）。在注冊 zkListener 時會返回對應的子節點注冊信息，並通知 listener（2.5）。

也就是說，訂閱時首先獲取該服務在 ZK 上全量注冊信息，之后消費者感知注冊信息變化，則是通過 zkListener 事件通知的方式。

3.4 NacosRegistry

NacosRegistry 注冊中心和 ZookeeperRegistry 類似，也是通過事件通知的方式感知服務注冊信息變化。

3.4.1 注冊

NacosRegistry 注冊時會將 URL 轉化為 Nacos 的實例對象 Instance，調用 registerInstance 進行注冊，deregisterInstance 取消注冊。

public void doRegister(URL url) {
    final String serviceName = getServiceName(url);
    final Instance instance = createInstance(url);
    execute(namingService -> namingService.registerInstance(serviceName, instance));
}

總結： NacosRegistry 注冊非常簡單，主要分析一下在 Nacos 上注冊的數據結構。

• serviceName：{category}:{serviceInterface}:{version}:{group}。其中 version 和 group 可以缺省，eg: providers:org.apache.dubbo.demo.DemoService::

• instance：這是 Nacos 的服務實例模型。

Nacos 注冊示例如下：

"providers:org.apache.dubbo.demo.DemoService::" -> {"enabled":true,"ephemeral":true,"healthy":true,"instanceHeartBeatInterval":5000,"instanceHeartBeatTimeOut":15000,"ip":"192.168.139.1","ipDeleteTimeout":30000,"metadata":{"side":"provider","methods":"sayHello","release":"","deprecated":"false","dubbo":"2.0.2","pid":"1128","interface":"org.apache.dubbo.demo.DemoService","generic":"false","path":"org.apache.dubbo.demo.DemoService","protocol":"dubbo","application":"dubbo-provider","dynamic":"true","category":"providers","anyhost":"true","bean.name":"org.apache.dubbo.demo.DemoService","register":"true","timestamp":"1570933206811"},"port":20880,"weight":1.0}

3.4.2 訂閱

訂閱時，首先獲取需要訂閱的服務名稱，和 ZK 一樣，也分為 Admin 和普通 serviceInterface 訂閱二種情況。

public void doSubscribe(final URL url, final NotifyListener listener) {
    Set<String> serviceNames = getServiceNames(url, listener);
    doSubscribe(url, listener, serviceNames);
}

// 訂閱指定的 serviceNames
private void doSubscribe(final URL url, final NotifyListener listener, 
                         final Set<String> serviceNames) {
    execute(namingService -> {
        for (String serviceName : serviceNames) {
            List<Instance> instances = namingService.getAllInstances(serviceName);
            // 通知 listener，將 Nacos Instance 適配成 Dubbo URL 后通知 listener
            notifySubscriber(url, listener, instances);
            // 注冊監聽器，感知服務注冊信息變化
            subscribeEventListener(serviceName, url, listener);
        }
    });
}

總結： 和 ZookeeperRegistry 類似，首先獲取對應服務名稱的服務實例，通過 notifySubscriber 通知 listener。之后服務感知也是通過 subscribeEventListener 事件機制。

private void subscribeEventListener(String serviceName, final URL url, 
		final NotifyListener listener) throws NacosException {
    if (!nacosListeners.containsKey(serviceName)) {
        EventListener eventListener = event -> {
            if (event instanceof NamingEvent) {
                NamingEvent e = (NamingEvent) event;
                // 服務注冊信息變化時通知 listener
                notifySubscriber(url, listener, e.getInstances());
            }
        };
        // 注冊EventListener
        namingService.subscribe(serviceName, eventListener);
        nacosListeners.put(serviceName, eventListener);
    }
}

4. 總結

Dubbo 對注冊中心進行了統一的抽象，核心接口是 RegistryService，其子類 AbstractRegistry 實現了緩存機制，FailbackRegistry 實現了重試機制。

ZookeeperRegistry、NacosRegistry 則具體等實現，則是完成具體的服務注冊和訂閱。注冊比較簡單，訂閱主要是通過事件機制，當注冊的服務發生變化時調用 notify(URL url, NotifyListener listener, List<URL> urls) 方法，更新內存和磁盤上本地緩存的注冊信息，並通知監聽者。

其中 RegistryDirectory 就是其中一個監聽者，會感知服務信息的變化，管理某個服務對應的所有注冊信息。

4.1 服務自省

我們知道 Dubbo 的注冊是以服務接口 serviceInterface 為單位進行注冊的，而大多數注冊中心的設計都是以服務實例為單位進行注冊的，如 Nacos、eureka、Spring Cloud 等。以服務實例進行注冊更接近雲原先，而且以服務接口為單位進行注冊，會造成注冊中心數據冗余，網絡通信壓力增大，減少注冊中心的吞吐量。

Dubbo 計划在 2.7.5 實現服務自省的功能，而 Spring Cloud alibaba-2.1.0 則已經完成了服務的自省。

圖5 Dubbo服務自省架構圖

注：本圖來源 小馬哥技術周報

---

每天用心記錄一點點。內容也許不重要，但習慣很重要！