Dubbo Consumer响应provider变化过程

2019-10-25 本文已影响0人晴天哥_王志

开篇

这篇文章用于分析Dubbo在Provider发生变化时Consumer感知变化并更新invoker的过程，这篇文章不会分析provider到invoker之间的转换过程，这部分过程会有单独的文章进行分析。

通过这篇文章，会理解Consumer感知zookeeper上Providers的变化，获取Provider的URL，转换成invoker的整个过程。变化包括Provider的新增或减少。

服务回调源码分析

ZookeeperRegistry在订阅过程中会针对消费service下的核心目录包括providers、configurators、routers进行监听,toCategoriesPath(url)返回需要监听的目录结构。
监听的目录在zk的节点包括
/dubbo/org.apache.dubbo.demo.DemoService/providers
/dubbo/org.apache.dubbo.demo.DemoService/routers
/dubbo/org.apache.dubbo.demo.DemoService/configurators
创建监听回调Listener对象，根据consumer的URL进行区分，listeners.putIfAbsent()内部通过lambda表达式创建Listener对象。
回调的Listener对象为：(parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)。
currentChilds为zookeeper监听目录子节点变化后目录返回的监听目录下所有子节点，针对providers目录而言，就是所有的provider的URL信息。
toUrlsWithEmpty()会针对返回的子节点currentChilds进行预处理，生成对应的URL列表。

public class ZookeeperRegistry extends FailbackRegistry {

    public void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            if (ANY_VALUE.equals(url.getServiceInterface())) {
                // 暂时不关注这部分代码
            } else {
                List<URL> urls = new ArrayList<>();
                // path格式如/dubbo/interface/provider
                for (String path : toCategoriesPath(url)) {
                    // 针对每个URL创建zk的监听回调
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) {
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                        listeners = zkListeners.get(url);
                    }
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) {
                        // 回调函数中核心的toUrlsWithEmpty重点关注
                        listeners.putIfAbsent(listener, 
                                 (parentPath, currentChilds) -> 
                                 ZookeeperRegistry.this.notify(url, 
                                 listener, toUrlsWithEmpty(url, parentPath, currentChilds)));

                        zkListener = listeners.get(listener);
                    }
                    zkClient.create(path, false);
                    // 针对path添加zk监听回调，感知path下子节点的变化
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) {
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                notify(url, listener, urls);
            }
        } catch (Throwable e) {
            
        }
    }
}

针对provider的List信息，会进行两步处理，toUrlsWithoutEmpty()和toUrlsWithEmpty()。
toUrlsWithoutEmpty()会针对providers为空的情况返回空列表，同时会将providers和consumer进行对比返回匹配consumer的方法。
toUrlsWithEmpty()会针对providers为空的情况生成empty://协议的URL对象。
在处理回调的notify()函数内部会针对这两种情况进行不同的逻辑判断。

    private List<URL> toUrlsWithEmpty(URL consumer, String path, List<String> providers) {
        List<URL> urls = toUrlsWithoutEmpty(consumer, providers);
        // 如果URL为空的情况就会新增一个empty协议的URL对象
        if (urls == null || urls.isEmpty()) {
            int i = path.lastIndexOf(PATH_SEPARATOR);
            String category = i < 0 ? path : path.substring(i + 1);
            URL empty = URLBuilder.from(consumer)
                    .setProtocol(EMPTY_PROTOCOL)
                    .addParameter(CATEGORY_KEY, category)
                    .build();
            urls.add(empty);
        }
        return urls;
    }

    private List<URL> toUrlsWithoutEmpty(URL consumer, List<String> providers) {
        List<URL> urls = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(providers)) {
            // 针对不同provider的URL进行consumer的匹配逻辑对比
            for (String provider : providers) {
                provider = URL.decode(provider);
                if (provider.contains(PROTOCOL_SEPARATOR)) {
                    URL url = URL.valueOf(provider);
                    if (UrlUtils.isMatch(consumer, url)) {
                        urls.add(url);
                    }
                }
            }
        }
        return urls;
    }

针对provider是否为空返回的urls信息，不为空的情况下为"dubbo://"，为空的情况下为"empty://"。

provider不为空情况下对应的URL内容：

dubbo://192.168.0.10:11111/org.apache.dubbo.demo.DemoService?anyhost=true
&application=dubbo-demo-api-provider&deprecated=false&dubbo=2.0.2&dynamic=true
&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello
&pid=3281&release=2.7.2&side=provider&timestamp=1571930584306


provider为空情况下对应的URL内容：

empty://192.168.0.10/org.apache.dubbo.demo.DemoService?application=dubbo-demo-api-consumer
&category=providers&dubbo=2.0.2&interface=org.apache.dubbo.demo.DemoService&lazy=false
&methods=sayHello&pid=3790&release=2.7.2&side=consumer&sticky=false&timestamp=1571931742294

notify()会根据urls当中的协议类型不同进行分组，包括configurator、provider、router等。
针对provider的情况调用refreshOverrideAndInvoker()更新invoker对象。

public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {

    public synchronized void notify(List<URL> urls) {
        // 根据URL的协议不同返回Map对象
        Map<String, List<URL>> categoryUrls = urls.stream()
                .filter(Objects::nonNull)
                .filter(this::isValidCategory)
                .filter(this::isNotCompatibleFor26x)
                .collect(Collectors.groupingBy(url -> {
                    if (UrlUtils.isConfigurator(url)) {
                        return CONFIGURATORS_CATEGORY;
                    } else if (UrlUtils.isRoute(url)) {
                        return ROUTERS_CATEGORY;
                    } else if (UrlUtils.isProvider(url)) {
                        return PROVIDERS_CATEGORY;
                    }
                    return "";
                }));

        List<URL> configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
        this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);

        List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList());
        toRouters(routerURLs).ifPresent(this::addRouters);

        // 更新provider的对象
        List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());

        refreshOverrideAndInvoker(providerURLs);
    }
}

refreshOverrideAndInvoker()内部会执行更新invoker的操作refreshInvoker()。
当provider为空的场景下，执行destroyAllInvokers()销毁所有的invoker对象。
当provider不为空的场景下，根据最新的invokerUrls生成invoker对象newUrlInvokerMap。
对比新旧invokers并执行destroyUnusedInvokers()销毁已经下线的invokers。

public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {


    private void refreshOverrideAndInvoker(List<URL> urls) {
        // mock zookeeper://xxx?mock=return null
        overrideDirectoryUrl();
        refreshInvoker(urls);
    }


    private void refreshInvoker(List<URL> invokerUrls) {
        Assert.notNull(invokerUrls, "invokerUrls should not be null");

        // 处理providers为空的情况
        if (invokerUrls.size() == 1
                && invokerUrls.get(0) != null
                && EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
            this.forbidden = true; // Forbid to access
            this.invokers = Collections.emptyList();
            routerChain.setInvokers(this.invokers);
            destroyAllInvokers(); // Close all invokers
        } else {
            // 处理providers不为空的情况
            this.forbidden = false; // Allow to access
            Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
            if (invokerUrls == Collections.<URL>emptyList()) {
                invokerUrls = new ArrayList<>();
            }
            if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
                invokerUrls.addAll(this.cachedInvokerUrls);
            } else {
                this.cachedInvokerUrls = new HashSet<>();
                this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
            }
            if (invokerUrls.isEmpty()) {
                return;
            }

            // 根据最新的URL生成invoker的Map对象
            Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map

            if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) {
                return;
            }

            List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
            routerChain.setInvokers(newInvokers);
            this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
            this.urlInvokerMap = newUrlInvokerMap;

            try {
                // 对比新旧invoker并销毁已下线的invoker
                destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
            } catch (Exception e) {
                logger.warn("destroyUnusedInvokers error. ", e);
            }
        }
    }
}

销毁的invoker对比逻辑是根据遍历旧的invoker对象，如果不在新的invoker对象列表里就认为该下线。
下线的逻辑是执行invoker.destroy()进行销毁即可。

    private void destroyUnusedInvokers(Map<String, Invoker<T>> oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {
        if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
            destroyAllInvokers();
            return;
        }
        // check deleted invoker
        List<String> deleted = null;
        if (oldUrlInvokerMap != null) {
            Collection<Invoker<T>> newInvokers = newUrlInvokerMap.values();
            for (Map.Entry<String, Invoker<T>> entry : oldUrlInvokerMap.entrySet()) {
                if (!newInvokers.contains(entry.getValue())) {
                    if (deleted == null) {
                        deleted = new ArrayList<>();
                    }
                    deleted.add(entry.getKey());
                }
            }
        }

        if (deleted != null) {
            for (String url : deleted) {
                if (url != null) {
                    Invoker<T> invoker = oldUrlInvokerMap.remove(url);
                    if (invoker != null) {
                        try {
                            invoker.destroy();
                            if (logger.isDebugEnabled()) {
                                logger.debug("destroy invoker[" + invoker.getUrl() + "] success. ");
                            }
                        } catch (Exception e) {
                            logger.warn("destroy invoker[" + invoker.getUrl() + "] failed. " + e.getMessage(), e);
                        }
                    }
                }
            }
        }
    }

Dubbo Consumer响应provider变化过程

开篇

服务回调源码分析

猜你喜欢

热点阅读