开篇
这篇文章的目的是尝试将dubbo的服务引用过程尽量描述的清楚些，主要核心在于注册consumer，订阅provider并生成reference的invoker对象。
另外，尝试通过zookeeper订阅过程回调描述清楚provider变更时consumer如何感知provider的增删操作。
注册中心类依赖图
说明：
- 关注ZookeeperRegistry、FailbackRegistry、AbastractRegistry的类关系。
- 关注上述三者的依赖关系由于理解consumer的注册过程。
consumer初始化过程时序图
说明：
- 1、consumer端在初始化bean后的afterPropertiesSet()方法中进行初始化动作。
- 2、consumer注册自身到zk的/dubbo/xxx_interface/consumer节点下面
- 3、consumer订阅zk的/dubbo/xxx_interface/provider|configor|routers服务
- 4、订阅节点有变化的时候通知变更
- 5、所有订阅的服务保存在RegistyDirectory目录中供发起RPC调用的时候cluster调用
整体流程图总结
说明：
- 以下内容参考自dubbo剖析：二 服务引用
- 1、引用配置及配置初始化
该部分以Spring配置及ReferenceBean为入口，主要在ReferenceConfig中进行。
ReferenceConfig依赖RegistryProtocol完成了 "服务引用者注册"、"服务提供者订阅"和"Invoker创建" 的工作；
ReferenceConfig依赖JavassistProxyFactory完成了 "代理对象生成" 的工作；
2、注册中心订阅 & Invoker生成与获取
该部分主要由RegistryDirectory和FailfastCluster实现。
通过ReferenceConfig调用RegistryDirectory的subscribe方法，触发了对服务提供者url的订阅及监听，在监听过程中RegistryDirectory借助DubboProtocol完成了Invoker的创建工作，并保存了服务引用url和Invoker的关系；
通过ReferenceConfig调用FailfastCluster的join方法，完成了对Invoker对象的获取；
3、生成代理对象
该部分主要由JavassistProxyFactory完成。
以ReferenceConfig调用JavassistProxyFactory的getProxy方法为入口，传入Invoker；
新创建了InvokerInvocationHandler，并使用dubbo自己的动态代理工具Proxy最终生成代理对象T ref；
dubbo 服务引用过程 - 阶段一
说明：
- 1、ReferenceBean的afterPropertiesSet()方法作为consumer的初始化入口。
- 2、afterPropertiesSet()内部调用getObject() -> get()执行ReferenceConfig的get()方法。
- 3、ReferenceConfig执行get() -> init() -> createProxy()进入consumer的代理创建过程。
- 4、createProxy()方法内部遍历注册中心生成consumer需要注册的注册中心URL地址。
- 5、关注refprotocol.refer方法，进入consumer引用provider的逻辑。
public class ReferenceBean<T> extends ReferenceConfig<T> implements
FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {
@SuppressWarnings({"unchecked"})
public void afterPropertiesSet() throws Exception {
// 省略相关代码
Boolean b = isInit();
if (b == null && getConsumer() != null) {
b = getConsumer().isInit();
}
if (b != null && b.booleanValue()) {
// 获取referenceBean入口
getObject();
}
}
public Object getObject() throws Exception {
return get();
}
}
public class ReferenceConfig<T> extends AbstractReferenceConfig {
public synchronized T get() {
if (ref == null) {
init();
}
return ref;
}
private void init() {
// 组装参数
Map<String, String> map = new HashMap<String, String>();
Map<Object, Object> attributes = new HashMap<Object, Object>();
map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
if (ConfigUtils.getPid() > 0) {
map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
}
// 创建服务引用的入口
ref = createProxy(map);
ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
}
private T createProxy(Map<String, String> map) {
if (isJvmRefer) {
// 省略相关代码
} else {
// user specified URL, could be peer-to-peer address,
// or register center's address.
// 处理直连的情况
if (url != null && url.length() > 0) {
String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (url.getPath() == null || url.getPath().length() == 0) {
url = url.setPath(interfaceName);
}
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else {
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else {
// assemble URL from register center's configuration
// 加载配置的所有注册中心，拼装成urls
List<URL> us = loadRegistries(false);
if (us != null && us.size() > 0) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if (monitorUrl != null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
// 这部分其实把注册中心和reference的信息进行了合并，后面的url是合并信息
urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
}
}
}
// 根据注册中心的数量选择走分支，这里一般情况走if分支。
if (urls.size() == 1) {
// 单注册中心或者直连的服务
invoker = refprotocol.refer(interfaceClass, urls.get(0));
} else {
// 多注册中心，遍历urls，调用refProtocol.refer创建远程的动态代理Invoker
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
// 多注册中心用最后一个注册中心的地址
for (URL url : urls) {
invokers.add(refprotocol.refer(interfaceClass, url));
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // use last registry url
}
}
// 多注册中心需要通过cluster选择一个
if (registryURL != null) { // registry url is available
// use AvailableCluster only when register's cluster is available
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
} else { // not a registry url
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
// 创建一个代理
return (T) proxyFactory.getProxy(invoker);
}
}dubbo 服务引用过程 - 阶段二
说明：
- 1、RegistryProtocol内部的doRefer()方法执行consumer注册和provider的订阅。
- 2、registry.register()负责注册consumer到zookeeper的对应节点路径上。
- 3、directory.subscribe()负责订阅provider/configurator/routers等节点路径信息。
public class RegistryProtocol implements Protocol {
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 从url的registryKey获取注册中心类型：zookeeper
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
// 从RegistryFactory获取注册器
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// 省略相关代码
// 关注走doRefer这部分逻辑
return doRefer(cluster, registry, type, url);
}
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);
// all attributes of REFER_KEY
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
// 构建subscribeUrl信息，主要拼接consumer:xxx的url地址
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
&& url.getParameter(Constants.REGISTER_KEY, true)) {
// 向注册中心注册服务消费者
registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
Constants.CHECK_KEY, String.valueOf(false)));
}
// 从注册中心订阅服务提供者（即引用的服务）
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
// 从invoker当中选择其中一个返回
Invoker invoker = cluster.join(directory);
// 注册消费者
ProviderConsumerRegTable.registerConsuemr(invoker, url, subscribeUrl, directory);
return invoker;
}
}dubbo 服务引用过程 - 阶段三
说明：
- 1、ZookeeperRegistry的实现consumer的对应的节点的注册。
- 2、zkClient.create(toUrlPath(url))负责创建zookeeper节点信息。
public abstract class AbstractRegistry implements Registry {
public void register(URL url) {
registered.add(url);
}
}
public abstract class FailbackRegistry extends AbstractRegistry {
public void register(URL url) {
if (destroyed.get()){
return;
}
super.register(url);
failedRegistered.remove(url);
failedUnregistered.remove(url);
try {
// 注册zookeeper节点上
doRegister(url);
} catch (Exception e) {
// 省略相关代码
}
}
protected abstract void doRegister(URL url);
}
public class ZookeeperRegistry extends FailbackRegistry {
protected void doRegister(URL url) {
try {
zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
} catch (Throwable e) {
// 省略相关代码
}
}
}dubbo 服务引用过程 - 阶段四
说明：
- 1、RegistryDirectory完成consumer对provider端的zk节点的订阅及回调处理。
- 2、RegistryDirectory的subscribe()方法调用FailbackRegistry.subscribe()方法。
- 3、FailbackRegistry.subscribe()内部调用ZookeeperRegistry.doSubscribe()方法。
4、ZookeeperRegistry.doSubscribe()方法内部遍历以下目录挨个进行订阅。
/dubbo/com.alibaba.dubbo.demo.DemoService/providers /dubbo/com.alibaba.dubbo.demo.DemoService/configurators /dubbo/com.alibaba.dubbo.demo.DemoService/routers- 5、new ChildListener()负责创建zk监听的回调函数，内部childChanged负责执行notify回调。
- 6、zkClient.addChildListener()负责获取path下所有节点信息，如providers目录下所有的provider的列表，即provider的URL。
- 7、FailbackRegistry.notify()方法用于针对providers/configurators/routers目录下的urls进行回调操作，用于初始化对应的类似invoker操作。
- 8、childChanged()方法内部的ZookeeperRegistry.this.notify()方法执行监听节点变化并进行重新初始化。
- 9、 FailbackRegistry.notify()内部调用listener.notify()回调RegistryDirectory.notify()方法，listener指代RegistryDirectory对象实例。
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
public void subscribe(URL url) {
//todo 关心这个对象的notify回调函数
setConsumerUrl(url);
registry.subscribe(url, this);
}
}
public abstract class FailbackRegistry extends AbstractRegistry {
public void subscribe(URL url, NotifyListener listener) {
if (destroyed.get()){
return;
}
super.subscribe(url, listener);
removeFailedSubscribed(url, listener);
try {
// Sending a subscription request to the server side
doSubscribe(url, listener);
} catch (Exception e) {
// 省略相关代码
}
}
protected abstract void doSubscribe(URL url, NotifyListener listener);
}
public class ZookeeperRegistry extends FailbackRegistry {
protected void doSubscribe(final URL url, final NotifyListener listener) {
try {
if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
// 省略相关代码
} else {
List<URL> urls = new ArrayList<URL>();
// path可以取以下的值
// /dubbo/com.alibaba.dubbo.demo.DemoService/providers
// /dubbo/com.alibaba.dubbo.demo.DemoService/configurators
// /dubbo/com.alibaba.dubbo.demo.DemoService/routers
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实例调用回调notify方法
ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
}
});
zkListener = listeners.get(listener);
}
zkClient.create(path, false);
// 获取path下的所有子节点并监听path路径
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
// 根据获取path路径下的子节点名称，就是provider的URL路径
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
// 通过回调将consumer:xxx的url路径及对应的
// providers/configurators/routers的路径进行回调
notify(url, listener, urls);
}
} catch (Throwable e) {
// 省略相关代码
}
}
}
public abstract class FailbackRegistry extends AbstractRegistry {
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");
}
try {
doNotify(url, listener, urls);
} catch (Exception t) {
// 省略相关代码
}
}
protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
super.notify(url, listener, urls);
}
}
public abstract class AbstractRegistry implements Registry {
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
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);
}
// 遍历providers/configurators/routers目录下的所有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是RegistryDirectory对象
listener.notify(categoryList);
}
}
}dubbo 服务引用过程 - 阶段五
说明：
- 1、RegistryDirectory.notify()方法调用refreshInvoker(invokerUrls)实现invoker的创建。
- 2、refreshInvoker()方法内部执行两个动作，分别创建新增的provider的invoker并下线删除的provider的invoker。
- 3、toInvokers()内部通过protocol.refer()方法创建provider的引用的invoker。
- 4、destroyUnusedInvokers()内部负责清理下线的provider的invoker。
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
public synchronized void notify(List<URL> urls) {
//todo step1 根据类别将urls分类
List<URL> invokerUrls = new ArrayList<URL>();
List<URL> routerUrls = new ArrayList<URL>();
List<URL> configuratorUrls = new ArrayList<URL>();
for (URL url : urls) {
String protocol = url.getProtocol();
String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
if (Constants.ROUTERS_CATEGORY.equals(category)
|| Constants.ROUTE_PROTOCOL.equals(protocol)) {
routerUrls.add(url);
} else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
|| Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
configuratorUrls.add(url);
} else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
invokerUrls.add(url);
} else {
logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
}
}
// configurators
if (configuratorUrls != null && configuratorUrls.size() > 0) {
this.configurators = toConfigurators(configuratorUrls);
}
// routers
if (routerUrls != null && routerUrls.size() > 0) {
List<Router> routers = toRouters(routerUrls);
if (routers != null) { // null - do nothing
setRouters(routers);
}
}
List<Configurator> localConfigurators = this.configurators; // local reference
// merge override parameters
this.overrideDirectoryUrl = directoryUrl;
if (localConfigurators != null && localConfigurators.size() > 0) {
for (Configurator configurator : localConfigurators) {
this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
}
}
// todo providers 内部的protocol.refer根据url创建远程代理Invoker
refreshInvoker(invokerUrls);
}
private void refreshInvoker(List<URL> invokerUrls) {
if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
&& Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
// 省略相关代码
} else {
this.forbidden = false; // Allow to access
Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
if (invokerUrls.size() == 0 && this.cachedInvokerUrls != null) {
invokerUrls.addAll(this.cachedInvokerUrls);
} else {
this.cachedInvokerUrls = new HashSet<URL>();
this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
}
if (invokerUrls.size() == 0) {
return;
}
//todo 根据url创建远程代理Invoker
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // Change method name to map Invoker Map
// state change
// If the calculation is wrong, it is not processed.
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls.toString()));
return;
}
this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
this.urlInvokerMap = newUrlInvokerMap;
try {
destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
} catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}
private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
String key = url.toFullString(); // The parameter urls are sorted
if (keys.contains(key)) { // Repeated url
continue;
}
keys.add(key);
// Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
if (invoker == null) { // Not in the cache, refer again
try {
// protocol.refer进行invoker创建的流程，待后续进行分析。
if (enabled) {
invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
}
} catch (Throwable t) {
}
if (invoker != null) { // Put new invoker in cache
newUrlInvokerMap.put(key, invoker);
}
} else {
newUrlInvokerMap.put(key, invoker);
}
}
keys.clear();
return newUrlInvokerMap;
}
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<String>();
}
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("destory invoker[" + invoker.getUrl() + "] success. ");
}
} catch (Exception e) {
logger.warn("destory invoker[" + invoker.getUrl() + "] faild. " + e.getMessage(), e);
}
}
}
}
}
}
}dubbo 服务引用过程 - 阶段六
说明：
- 1、consumer端的RegistryDirectory维持了provider的invoker信息。
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
private static final Logger logger = LoggerFactory.getLogger(RegistryDirectory.class);
private static final Cluster cluster = ExtensionLoader.getExtensionLoader(Cluster.class).getAdaptiveExtension();
private static final RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getAdaptiveExtension();
private static final ConfiguratorFactory configuratorFactory = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class).getAdaptiveExtension();
private final String serviceKey; // Initialization at construction time, assertion not null
private final Class<T> serviceType; // Initialization at construction time, assertion not null
private final Map<String, String> queryMap; // Initialization at construction time, assertion not null
private final URL directoryUrl; // Initialization at construction time, assertion not null, and always assign non null value
private final String[] serviceMethods;
private final boolean multiGroup;
private Protocol protocol; // Initialization at the time of injection, the assertion is not null
private Registry registry; // Initialization at the time of injection, the assertion is not null
private volatile boolean forbidden = false;
private volatile URL overrideDirectoryUrl; // Initialization at construction time, assertion not null, and always assign non null value
/**
* override rules
* Priority: override>-D>consumer>provider
* Rule one: for a certain provider <ip:port,timeout=100>
* Rule two: for all providers <* ,timeout=5000>
*/
private volatile List<Configurator> configurators; // The initial value is null and the midway may be assigned to null, please use the local variable reference
// Map<url, Invoker> cache service url to invoker mapping.
private volatile Map<String, Invoker<T>> urlInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference
// Map<methodName, Invoker> cache service method to invokers mapping.
private volatile Map<String, List<Invoker<T>>> methodInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference
// Set<invokerUrls> cache invokeUrls to invokers mapping.
private volatile Set<URL> cachedInvokerUrls; // The initial value is null and the midway may be assigned to null, please use the local variable reference
}
