开篇
这篇文章用于分析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×tamp=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×tamp=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);
}
}
}
}
}
}
网友评论