天天看點
傳回

Dubbo Admin實作原理

開篇

  • 這篇文章主要用于講解清楚Dubbo Admin的監控資料的來源，如何通過zookeeper作為注冊中心來擷取實際運作中的資料進行服務治理。
  • 這篇文章主要側重于資料方面的擷取包括consumers, configurators, routers, providers，具體如何進行服務治理後面文章會具體進行分析。
  • 文章是基于dubbo-2.6.0的版本進行分析。

注冊節點 

[zk] ls /dubbo/com.alibaba.dubbo.demo.DemoService
[consumers, configurators, routers, providers]
Dubbo Admin實作原理
  • Dubbo的服務樹如上圖所示，以服務如com.foo.BarService作為第一層節點Service層。
  • Service層包含Type層，包括consumers, configurators, routers, providers，監控資料主要是指Type層子節點的資料監控。

實作過程概述

  • 1.依次按照/dubbo，/dubbo/service，/dubbo/service/type的順序進行發現并最終訂閱/dubbo/service/type的節點。
  • 2.首次擷取所有/dubbo/service/type的所有子節點并儲存導資料結構registryCache當中。
  • 3.後續/dubbo/service/type的節點變更通過zookeeper事件通知機制更新到registryCache當中。

  • 4.registryCache儲存了zookeeper上Dubbo服務節點上的所有資訊，按照

    ConcurrentMap>>的資料結構進行儲存，其中category包含providers,consumers,routers,configurators。

實作過程源碼分析

Dubbo Admin實作原理
  • dubbo-admin的目錄結構如上圖，RegistryServerSync作為同步資料的核心類。
  • RegistryServerSync的資料用于同步并監聽zookeeper中關于Dubbo服務的資料。

RegistryServerSync的bean定義 

<dubbo:application name="dubbo-admin"/>

<dubbo:registry address="${dubbo.registry.address}" check="false" file="false"/>

<dubbo:reference id="registryService" interface="com.alibaba.dubbo.registry.RegistryService" check="false"/>

<bean id="configService" class="com.alibaba.dubbo.governance.service.impl.ConfigServiceImpl"/>
<bean id="consumerService" class="com.alibaba.dubbo.governance.service.impl.ConsumerServiceImpl"/>
<bean id="overrideService" class="com.alibaba.dubbo.governance.service.impl.OverrideServiceImpl"/>
<bean id="ownerService" class="com.alibaba.dubbo.governance.service.impl.OwnerServiceImpl"/>
<bean id="providerService" class="com.alibaba.dubbo.governance.service.impl.ProviderServiceImpl"/>
<bean id="routeService" class="com.alibaba.dubbo.governance.service.impl.RouteServiceImpl"/>
<bean id="userService" class="com.alibaba.dubbo.governance.service.impl.UserServiceImpl">
    <property name="rootPassword" value="${dubbo.admin.root.password}"/>
    <property name="guestPassword" value="${dubbo.admin.guest.password}"/>
</bean>

<bean id="governanceCache" class="com.alibaba.dubbo.governance.sync.RegistryServerSync"/>
  • RegistryServerSync的定義在META-INF/spring/dubbo-admin.xml當中。

RegistryServerSync初始化訂閱過程 

public class RegistryServerSync implements InitializingBean, DisposableBean, NotifyListener {

    private static final Logger logger = LoggerFactory.getLogger(RegistryServerSync.class);

    // admin://192.168.1.5?category=providers,consumers,routers,configurators&check=false&classifier=*&enabled=*&group=*&interface=*&version=*
    private static final URL SUBSCRIBE = new URL(Constants.ADMIN_PROTOCOL, NetUtils.getLocalHost(), 0, "",
            Constants.INTERFACE_KEY, Constants.ANY_VALUE,
            Constants.GROUP_KEY, Constants.ANY_VALUE,
            Constants.VERSION_KEY, Constants.ANY_VALUE,
            Constants.CLASSIFIER_KEY, Constants.ANY_VALUE,
            Constants.CATEGORY_KEY, Constants.PROVIDERS_CATEGORY + ","
            + Constants.CONSUMERS_CATEGORY + ","
            + Constants.ROUTERS_CATEGORY + ","
            + Constants.CONFIGURATORS_CATEGORY,
            Constants.ENABLED_KEY, Constants.ANY_VALUE,
            Constants.CHECK_KEY, String.valueOf(false));

    private static final AtomicLong ID = new AtomicLong();

    /**
     * Make sure ID never changed when the same url notified many times
     */
    private final ConcurrentHashMap<String, Long> URL_IDS_MAPPER = new ConcurrentHashMap<String, Long>();

    // ConcurrentMap<category, ConcurrentMap<servicename, Map<Long, URL>>>
    private final ConcurrentMap<String, ConcurrentMap<String, Map<Long, URL>>> registryCache = new ConcurrentHashMap<String, ConcurrentMap<String, Map<Long, URL>>>();
    @Autowired
    private RegistryService registryService;

    public ConcurrentMap<String, ConcurrentMap<String, Map<Long, URL>>> getRegistryCache() {
        return registryCache;
    }

    public void afterPropertiesSet() throws Exception {
        logger.info("Init Dubbo Admin Sync Cache...");
        registryService.subscribe(SUBSCRIBE, this);
    }
}
Dubbo Admin實作原理
  • RegistryServerSync在afterPropertiesSet()方法内部執行訂閱操作。
  • SUBSCRIBE變量中category=providers,consumers,routers,configurators表示我們需要訂閱service下的這4類節點，其中routers和configurators就是跟服務治理相關。
  • SUBSCRIBE的check=false&classifier=&enabled=&group=&interface=&version=*等變量都是通用符*，這裡我們關注下interface=*的變量，後面執行實際訂閱會根據該變量做判斷。

RegistryServerSync執行訂閱過程

  • 參見ZookeeperRegistry#doSubscribe 
protected void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            // admin訂閱過程中，首次傳入參數interface的值為*，是以走的這個分支
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
                String root = toRootPath();
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    listeners.putIfAbsent(listener, new ChildListener() {
                        public void childChanged(String parentPath, List<String> currentChilds) {
                            for (String child : currentChilds) {
                                child = URL.decode(child);
                                if (!anyServices.contains(child)) {
                                    anyServices.add(child);
                                    subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child,
                                            Constants.CHECK_KEY, String.valueOf(false)), listener);
                                }
                            }
                        }
                    });
                    zkListener = listeners.get(listener);
                }
                zkClient.create(root, false);
                // 擷取 /dubbo目錄的子節點，傳回service層所有service接口
                // 如/dubbo/com.alibaba.dubbo.demo.DemoServiceEcho
                List<String> services = zkClient.addChildListener(root, zkListener);
                if (services != null && services.size() > 0) {
                    for (String service : services) {
                        service = URL.decode(service);
                        anyServices.add(service);
                        // 針對每個service進行subscribe操作
                        subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
                                Constants.CHECK_KEY, String.valueOf(false)), listener);
                    }
                }
            } else {
                List<URL> urls = new ArrayList<URL>();
                // 擷取service下的category，consumers, configurators, routers, providers
                // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/providers
                // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/consumers
                // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/routers
                // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/configurators
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) {
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                        listeners = zkListeners.get(url);
                    }
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) {
                        listeners.putIfAbsent(listener, new ChildListener() {
                            public void childChanged(String parentPath, List<String> currentChilds) {
                                ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                            }
                        });
                        zkListener = listeners.get(listener);
                    }
                    zkClient.create(path, false);
                    // 針對每個service的監聽以下節點
                    // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/providers
                    // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/consumers
                    // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/routers
                    // /dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/configurators
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) {
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                // urls為某個服務下所有的子節點内容
                // dubbo://172.17.32.8:20880/com.alibaba.dubbo.demo.DemoServiceEcho?anyhost=true&application=demo-provider&bean.name=com.alibaba.dubbo.demo.DemoServiceEcho&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoServiceEcho&methods=sayHello&pid=79153&side=provider&timestamp=1577783208090
                // empty://192.168.1.5/com.alibaba.dubbo.demo.DemoServiceEcho?category=consumers&check=false&classifier=*&enabled=*&group=*&interface=com.alibaba.dubbo.demo.DemoServiceEcho&version=*
                // route://0.0.0.0/com.alibaba.dubbo.demo.DemoServiceEcho?category=routers&dynamic=false&enabled=true&force=true&name=com.alibaba.dubbo.demo.DemoServiceEcho blackwhitelist&priority=0&router=condition&rule=consumer.host+%3D+1.1.1.1+%3D%3E+false&runtime=false
                // override://192.168.1.5/com.alibaba.dubbo.demo.DemoServiceEcho?category=configurators&dynamic=false&enabled=true&weight=12
                // override://192.168.1.5/com.alibaba.dubbo.demo.DemoServiceEcho?category=configurators&dynamic=false&enabled=true&weight=12
                // override://1.1.1.1/com.alibaba.dubbo.demo.DemoServiceEcho?category=configurators&dynamic=false&enabled=true&weight=13
                notify(url, listener, urls);
            }
        } catch (Throwable e) {
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }
  • 訂閱的過程是一個逐層發現并訂閱的過程。
  • 擷取根節點/dubbo下的所有service節點集合，如/dubbo/com.alibaba.dubbo.demo.DemoServiceEcho。
  • 針對每個service節點擷取該節點的所有子節點(描述友善用type表示)，包括providers,consumers,routers,configurators。
  • 針對每個type如/dubbo/com.alibaba.dubbo.demo.DemoServiceEcho/providers進行監聽。
  • 首次啟動會擷取每個type節點下的所有子節點進行第一輪初始化過程，後續的變更都是通過type節點本身的監聽回調進行實作。
  • 首次啟動會把每個service下所有的type下的所有的子節點合并成urls後進行notify動作。

RegistryServerSync回調分組過程

  • 參見AbstractRegistry#notify 
public abstract class AbstractRegistry implements Registry {

    protected void notify(URL url, NotifyListener listener, List<URL> urls) {
        if (url == null) {
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) {
            throw new IllegalArgumentException("notify listener == null");
        }
        if ((urls == null || urls.size() == 0)
                && !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);
        }
        // 根據category進行分組，包括consumers, configurators, routers, providers
        Map<String, List<URL>> result = new HashMap<String, List<URL>>();
        for (URL u : urls) {
            if (UrlUtils.isMatch(url, u)) {
                String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
                List<URL> categoryList = result.get(category);
                if (categoryList == null) {
                    categoryList = new ArrayList<URL>();
                    result.put(category, categoryList);
                }
                categoryList.add(u);
            }
        }
        if (result.size() == 0) {
            return;
        }
        Map<String, List<URL>> categoryNotified = notified.get(url);
        if (categoryNotified == null) {
            notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
            categoryNotified = notified.get(url);
        }
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            saveProperties(url);
            // 按照分組的結果進行回調通知
            listener.notify(categoryList);
        }
    }
}
  • RegistryServerSync的回調過程中根據category進行分組，分組包括consumers, configurators, routers, providers。
  • 依次針對分組後的結果進行回調通知，執行RegistryServerSync的notify動作。

RegistryServerSync儲存回調結果

  • 參見RegistryServerSync#notify 
public class RegistryServerSync implements InitializingBean, DisposableBean, NotifyListener {
    /**
     * Make sure ID never changed when the same url notified many times
     */
    private final ConcurrentHashMap<String, Long> URL_IDS_MAPPER = new ConcurrentHashMap<String, Long>();

    // ConcurrentMap<category, ConcurrentMap<servicename, Map<Long, URL>>>
    // servicename=groupName/serviceName:versionNum
    private final ConcurrentMap<String, ConcurrentMap<String, Map<Long, URL>>> registryCache = new ConcurrentHashMap<String, ConcurrentMap<String, Map<Long, URL>>>();


    public void notify(List<URL> urls) {
        if (urls == null || urls.isEmpty()) {
            return;
        }
        // Map<category, Map<servicename, Map<Long, URL>>>
        final Map<String, Map<String, Map<Long, URL>>> categories = new HashMap<String, Map<String, Map<Long, URL>>>();
        String interfaceName = null;
        for (URL url : urls) {
            String category = url.getParameter(Constants.CATEGORY_KEY, Constants.PROVIDERS_CATEGORY);
            // 針對empty的情況，移出已經消失的服務
            if (Constants.EMPTY_PROTOCOL.equalsIgnoreCase(url.getProtocol())) { // NOTE: group and version in empty protocol is *
                ConcurrentMap<String, Map<Long, URL>> services = registryCache.get(category);
                if (services != null) {
                    String group = url.getParameter(Constants.GROUP_KEY);
                    String version = url.getParameter(Constants.VERSION_KEY);
                    // NOTE: group and version in empty protocol is *
                    if (!Constants.ANY_VALUE.equals(group) && !Constants.ANY_VALUE.equals(version)) {
                        services.remove(url.getServiceKey());
                    } else {
                        for (Map.Entry<String, Map<Long, URL>> serviceEntry : services.entrySet()) {
                            String service = serviceEntry.getKey();
                            if (Tool.getInterface(service).equals(url.getServiceInterface())
                                    && (Constants.ANY_VALUE.equals(group) || StringUtils.isEquals(group, Tool.getGroup(service)))
                                    && (Constants.ANY_VALUE.equals(version) || StringUtils.isEquals(version, Tool.getVersion(service)))) {
                                services.remove(service);
                            }
                        }
                    }
                }
            } else {
                // 添加服務到全局的registryCache變量當中
                if (StringUtils.isEmpty(interfaceName)) {
                    interfaceName = url.getServiceInterface();
                }
                // 用于儲存局部變量的categories
                Map<String, Map<Long, URL>> services = categories.get(category);
                if (services == null) {
                    services = new HashMap<String, Map<Long, URL>>();
                    categories.put(category, services);
                }
                // service=groupName/interfaceName:versionNum
                String service = url.getServiceKey();
                Map<Long, URL> ids = services.get(service);
                if (ids == null) {
                    ids = new HashMap<Long, URL>();
                    services.put(service, ids);
                }

                // Make sure we use the same ID for the same URL
                if (URL_IDS_MAPPER.containsKey(url.toFullString())) {
                    ids.put(URL_IDS_MAPPER.get(url.toFullString()), url);
                } else {
                    long currentId = ID.incrementAndGet();
                    ids.put(currentId, url);
                    URL_IDS_MAPPER.putIfAbsent(url.toFullString(), currentId);
                }
            }
        }
        if (categories.size() == 0) {
            return;
        }

        // 本次category對應的資料不為空進行添加動作
        for (Map.Entry<String, Map<String, Map<Long, URL>>> categoryEntry : categories.entrySet()) {
            String category = categoryEntry.getKey();
            ConcurrentMap<String, Map<Long, URL>> services = registryCache.get(category);
            if (services == null) {
                services = new ConcurrentHashMap<String, Map<Long, URL>>();
                registryCache.put(category, services);
            } else {
                Set<String> keys = new HashSet<String>(services.keySet());
                // 移出已經不存在的資料，同一個接口但是不在本次的回調資料當中
                for (String key : keys) {
                    if (Tool.getInterface(key).equals(interfaceName) && !categoryEntry.getValue().entrySet().contains(key)) {
                        services.remove(key);
                    }
                }
            }
            // 用最新的資料進行覆寫
            services.putAll(categoryEntry.getValue());
        }
    }
}
  • RegistryServerSync的notify()方法的urls參數是/dubbo/service/type下的子節點。
  • Dubbo的服務樹的資料最終通過ConcurrentMap>>的資料結構進行儲存，儲存的對象為registryCache，servicename=groupName/serviceName:versionNum的格式儲存。
  • registryCache儲存的Zookeeper上的Dubbo的服務節點内容。所有的服務治理操作依據的資料都在registryCache當中，任何服務節點的變更都會導緻registryCache的資料更新。
  • registryCache是整個服務治理的資料核心。
監控 Dubbo java 應用服務中間件 spring dubbo實作原理 dubbo admin安裝 dubbo admin原因 dubbo admin報錯 dubbo admin釋出
上一篇: 第 425 期 Python 周刊
下一篇: 成都音視訊技術沙龍邀你參加

繼續閱讀