开篇
整个Dubbo Consumer的引用过程比较复杂，这部分的文章会比较多，这篇文章的目的是描述Consumer的订阅过程，侧重于Consumer发现Provider的URL并生成对应的invoker的过程。
在这篇文章中，主要分为两个部分讲解，第一部分RegistryProtocol的refer过程侧重于描述RegistryProtocol的流程，ZookeeperRegistry的subscribe过程侧重于服务订阅过程。
RegistryProtocol的refer过程
- RegistryProtocol调用refer() => doRefer()方法。
- RegistryProtocol的registry是ZookeeperRegistry对象。
public class RegistryProtocol implements Protocol {
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?
// application=dubbo-demo-api-consumer&dubbo=2.0.2&pid=58968
// &refer=application=dubbo-demo-api-consumer&dubbo=2.0.2
// &interface=org.apache.dubbo.demo.DemoService
// &lazy=false&methods=sayHello&pid=58968
// ®ister.ip=172.17.32.176&side=consumer&sticky=false
// ×tamp=1571824631224®istry=zookeeper×tamp=1571824632206
url = URLBuilder.from(url)
.setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
.removeParameter(REGISTRY_KEY)
.build();
// registry为ZookeeperRegistry对象
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
String group = qs.get(GROUP_KEY);
if (group != null && group.length() > 0) {
if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
return doRefer(getMergeableCluster(), registry, type, url);
}
}
// 执行doRefer动作
return doRefer(cluster, registry, type, url);
}- RegistryProtocol的doRefer()执行一系列核心步骤，在代码中都备有注释。
- 创建RegistryDirectory对象。
- 生成服务消费者链接并在consumers目录下新增节点注册服务消费者。
- 创建路由规则链。
- 订阅 providers、configurators、routers 等节点数据。
- 将多个服务提供者合并为一个invoker。
- 重点关注订阅 providers、configurators、routers 等节点数据的流程
public class RegistryProtocol implements Protocol {
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// 创建RegistryDirectory实例
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
// 设置注册中心
directory.setRegistry(registry);
// 设置协议
directory.setProtocol(protocol);
// 所有属性放到map中
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
// 生成服务消费者链接
URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
// 注册服务消费者，在 consumers 目录下新节点
if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
registry.register(directory.getRegisteredConsumerUrl());
}
// 创建路由规则链
directory.buildRouterChain(subscribeUrl);
// 订阅 providers、configurators、routers 等节点数据
directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
// 一个注册中心可能有多个服务提供者，因此这里需要将多个服务提供者合并为一个，生成一个invoker
Invoker invoker = cluster.join(directory);
// 在服务提供者处注册消费者
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
return invoker;
}
}- RegistryDirectory中的registry是ZookeeperRegistry对象。
- ZookeeperRegistry中subscribe(url, this)方法中将RegistryDirectory对象作为NotifyListener参数。
- 执行ZookeeperRegistry对象的subscribe()方法，至此进入了服务订阅过程。
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
public void subscribe(URL url) {
setConsumerUrl(url);
CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this);
serviceConfigurationListener = new ReferenceConfigurationListener(this, url);
// registry为ZookeeperRegistry对象。
registry.subscribe(url, this);
}
}ZookeeperRegistry的subscribe过程
- ZookeeperRegistry的整个subscribe过程涉及到ZookeeperRegistry、FailbackRegistry 、AbstractRegistry这几个类，整个订阅过程如下图所示。
- 从图中可以发现父类FailbackRegistry作为执行subscribe()的入口，真正执行的是ZookeeperRegistry的doSubscribe()方法。
- ZookeeperRegistry、FailbackRegistry 、AbstractRegistry三者的类关系如下图，涉及到父类子类具体实现方法的调用。
- 从ZookeeperRegistry订阅时序图可以看出来订阅首先执行FailbackRegistry的subscribe()方法。
- FailbackRegistry的subscribe()方法调用子类的ZookeeperRegistry的doSubscribe()方法。
public abstract class FailbackRegistry extends AbstractRegistry {
public void subscribe(URL url, NotifyListener listener) {
super.subscribe(url, listener);
removeFailedSubscribed(url, listener);
try {
// 执行订阅操作
doSubscribe(url, listener);
} catch (Exception e) {
}
}
}- ZookeeperRegistry的doSubscribe()主要执行订阅操作，核心和zookeeper的特性比较相关。
- 获取待订阅的path信息，包括providers,configurators,routers三类路径。
- 针对每个待订阅的path，会针对path增加children维度的监听。
- 针对每个待订阅的path，会一次性获取children的变量并进入notify()流程。
- zkListeners先根据consumer的URL维度 和 listener维度唯一确定Zookeeper子节点监听器ChildListener对象。
- ZookeeperRegistry的path的子节点监听器ChildListener回调中执行的ZookeeperRegistry.this.notify()方法。
- 针对处理path下的children的URL路径会通过toUrlsWithEmpty()方法进行匹配，获取匹配的URL进行处理。
- 首次获取及后续回调都是执行ZookeeperRegistry.this.notify()方法。
- notify()的urls是指订阅path下所有符合要求的urls，通过toUrlsWithEmpty()进行匹配。
public class ZookeeperRegistry extends FailbackRegistry {
// 变量url的值
// consumer://172.17.32.176/org.apache.dubbo.demo.DemoService?
// application=dubbo-demo-api-consumer&category=providers,configurators,routers&dubbo=2.0.2
// &interface=org.apache.dubbo.demo.DemoService&lazy=false&methods=sayHello
// &pid=58968&side=consumer&sticky=false×tamp=1571824631224
public void doSubscribe(final URL url, final NotifyListener listener) {
try {
if (ANY_VALUE.equals(url.getServiceInterface())) {
// 暂时不关注这部分逻辑
} else {
List<URL> urls = new ArrayList<>();
// 处理providers、configurators、routers等路径
// /dubbo/org.apache.dubbo.demo.DemoService/providers
// /dubbo/org.apache.dubbo.demo.DemoService/configurators
// /dubbo/org.apache.dubbo.demo.DemoService/routers
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);
if (zkListener == null) {
// 创建zk节点变化回调监听器
listeners.putIfAbsent(listener,
(parentPath, currentChilds) -> ZookeeperRegistry.this.notify(
url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));
zkListener = listeners.get(listener);
}
// 创建path对应的节点
zkClient.create(path, false);
// 添加path下的children的监听
List<String> children = zkClient.addChildListener(path, zkListener);
// 处理path下的children
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
// 通知回调notify动作
notify(url, listener, urls);
}
} catch (Throwable e) {
}
}
}- ZookeeperRegistry.notify()方法会调用父类FailbackRegistry.notify()方法进而执行doNotify()方法。
- FailbackRegistry.doNotify()方法会调用父类AbstractRegistry.notify()。
public abstract class FailbackRegistry extends AbstractRegistry {
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
try {
doNotify(url, listener, urls);
} catch (Exception t) {
addFailedNotified(url, listener, urls);
}
}
protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
super.notify(url, listener, urls);
}
}- AbstractRegistry的notify()的urls是指订阅path下所有符合要求的urls。
- 将所有的urls按照url的category包括(providers,configurators,routers)三类。
- 将所有的urls按照分组进行回调处理。
- 回调的listener为RegistryDirectory对象，实现了NotifyListener接口。
public abstract class AbstractRegistry implements Registry {
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
Map<String, List<URL>> result = new HashMap<>();
for (URL u : urls) {
// 符合要求的URL按照category作为分组key的Map。
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();
// 同一类category的URL进行回调，譬如providers的URL一并进行回调。
List<URL> categoryList = entry.getValue();
categoryNotified.put(category, categoryList);
// 调用监听回调
listener.notify(categoryList);
// 保存URL信息
saveProperties(url);
}
}
}RegistryDirectory的notify过程
- 按照configurators、routers、providers组装成List对象。
- 针对configurators执行toConfigurators()动作。
- 针对routers执行toRouters()动作。
- 针对providers执行refreshOverrideAndInvoker()动作。
- 每个细节后面分篇幅继续讲解。
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
public synchronized void notify(List<URL> urls) {
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);
// providers
List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());
refreshOverrideAndInvoker(providerURLs);
}
}
