dubbo服务调用源码

发表于 2026-5-20 07:51:46

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要登录才可以下载或查看，没有账号？立即注册

×

调用过程

dubbo的服务调用方是在xml设置了类似于的设置，意味着后续在spring中通过 getBean('productService') 就可以获取到远程署理对象。dubbo:reference 本身映射成为的bean是ReferenceBean，其会存储整个dubbo须要的各种信息，比方控制中央的注册所在，服务端的详细IO和端口等。

如上图就是ReferenceBean的类图，根据以往对spring的学习相识，有如下总结：

很清楚的熟悉到其是一个工厂Bean，后续须要getObject方法得到真正的对象（其着实这里不看源码，我们就应该能猜到通例做法是通过动态署理天生interface="com.jwfy.dubbo.product.ProductService中接口对应的proxy对象），假如想获取ReferenceBean对象本身，则须要利用getBean("&productService")
通过InitializingBean的afterPropertiesSet 方法去为当前的bean注入注册中央、均衡负责的方式、利用的协议等属性数据。

在这里天生署理对象是通过Java的动态署理方式，由于指明的是接口，（spring中默认的是假如通过接口天生署理对象，是利用JDK动态署理，否则是利用cglib），那么根据对动态署理知识点的相识，InvocationHandler肯定是跑不掉的，通过invoke去调用实行函数的。
天生ReferenceBean的署理对象

在getObject方法为入口打断点，末了可以追踪到ReferenceConfig类的createProxy方法为真正的天生署理对象的操纵。

private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp", "localhost", 0, map);
// 这里的临时url可以是 temp://localhost?application=dubbo-consume&default.check=false
//&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=15813&side=consumer&timestamp=1525921242794
final boolean isJvmRefer;
if (isInjvm() == null) {
if (url != null && url.length() > 0) {
// 指定URL的情况下，不做本地引用
isJvmRefer = false;
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
// 默认情况下如果本地有服务暴露，则引用本地服务.
isJvmRefer = true;
} else {
isJvmRefer = false;
}
} else {
isJvmRefer = isInjvm().booleanValue();
}
if (isJvmRefer) {
// 如果是本地的服务
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
invoker = refprotocol.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
} else {
if (url != null && url.length() > 0) {
// 用户指定URL，指定的URL可能是对点对直连地址，也可能是注册中心URL
// 用户可以在dubbo:reference 中直接配置url参数，配置了就直接连接
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 { // 只能是通过注册中心配置拼装URL
List<URL> us = loadRegistries(false);
// 这个false的值含义是非服务提供方，获取到连接注册中心的配置
// 切记！！！不是获取服务提供方的url属性，而是注册中心的配置
// 生成的url是类似于registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?
// application=dubbo-consume&client=zkclient&dubbo=2.5.3&group=dubbo-demo&owner=jwfy&pid=1527&registry=zookeeper&timestamp=1525943117155
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()));
}
urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
}
}
if (urls == null || urls.size() == 0) {
// 没有找到一个可用的连接到注册中心的数据
throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address="..." /> to your spring config.");
}
}
// 在这里interfaceClass说的是 ==> com.jwfy.dubbo.product.ProductService
// urls还是register协议
// 总之下面这段代码是非常关键的，后面再补充
if (urls.size() == 1) {
invoker = refprotocol.refer(interfaceClass, urls.get(0));
// refprotocol就是包装了两次的RegistryProtocol，直接生成对应的invoker
} else {
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
// 在这里能够猜到其中一定有向注册中心订阅获取所有的服务提供方的信息
// 并把获取到的信息返回生成一个invoker对象
invokers.add(refprotocol.refer(interfaceClass, url));
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // 用了最后一个registry url
}
}
if (registryURL != null) { // 有注册中心协议的URL
// 对有注册中心的Cluster 只用 AvailableCluster
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
} else { // 不是注册中心的URL
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
Boolean c = check;
if (c == null && consumer != null) {
c = consumer.isCheck();
}
if (c == null) {
c = true; // default true
}
if (c && ! invoker.isAvailable()) {
throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
// 创建服务代理
// 上面已经说了proxyFactory是StubProxyFactoryWrapper包装了JavassistProxyFactory类
return (T) proxyFactory.getProxy(invoker);
}

复制代码

JavassistProxyFactory 类

public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}

复制代码

天生Invoker

Invoker是对可实行对象的引用，须要明白可引用的详细位置（服务端的明白IP和端口等信息）
如今已经拿到了当前服务的注册中央的设置，那么接下来就须要毗连到注册中央，并获取到可以调用的呆板情况（实际开辟中，分布式体系根本上都存在多个呆板信息），并组合成为须要的invoker。
下面这个代码段就是生产详细的Invoke操纵，接下来好好分析一下

invoker = refprotocol.refer(interfaceClass, urls.get(0));
// refprotocol就是包装了两次的RegistryProtocol,直接生成对于的invoker
// url是registry开头的协议，在参数中包含了消息等协议和其参数
// interfaceClass是接口类

复制代码

在分析源码之前，假如换成我们，我们会完成什么操纵呢？

向注册中央订阅斲丧者，如许通过dubbo-admin就可以观察现有的生产者和斲丧者
从注册中央获取生产者的信息
类似均衡负责的操纵，选择符合的生产者
直连生产者，获取效果

RegistryProtocol 类

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
// 替换协议，一般情况下都是替换成为zookeeper
Registry registry = registryFactory.getRegistry(url);
// 这一步是获取到注册中心的配置信息，这里和服务暴露的操作一致
// 其中的key是zookeeper://127.0.0.1:2182/dubbo-demo/com.alibaba.dubbo.registry.RegistryService
// 那么同一个jvm内的key其实都是一致的，所以其连接到控制中心的数据也一致
// 如果没有连接，则需要创建一个新的链接实体，其中会把当前的信息存储到文件中
// 也会有对应的监听者等，然后真正的调用连接zk操作，确保zk是真实存在的
if (RegistryService.class.equals(type)) {
// 如果类型是RegistryService，则使用getInvoke操作直接拼接生成一个invoke对象
// 这个操作和服务提供方生成invoker的方式一致
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
// 这个qs包含了需要生成invoker的所有参数，例如接口名称，函数名，所述范畴（消费者）等信息
String group = qs.get(Constants.GROUP_KEY);
// 查看分组信息
if (group != null && group.length() > 0 ) {
if ( ( Constants.COMMA_SPLIT_PATTERN.split( group ) ).length > 1
|| "*".equals( group ) ) {
return doRefer( getMergeableCluster(), registry, type, url );
}
}
// 看有分组和没分组，其实就是集群cluster不一样
return doRefer(cluster, registry, type, url);
}
private Cluster getMergeableCluster() {
return ExtensionLoader.getExtensionLoader(Cluster.class).getExtension("mergeable");
// 生成MergeableCluster 集群
// 如下代码Cluster$Adpative类则就是cluster，默认的是为FailoverCluster集群
}

复制代码

Cluster$Adpative 类

package com.alibaba.dubbo.rpc.cluster;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class Cluster$Adpative implements com.alibaba.dubbo.rpc.cluster.Cluster {
public com.alibaba.dubbo.rpc.Invoker join(
com.alibaba.dubbo.rpc.cluster.Directory arg0)
throws com.alibaba.dubbo.rpc.cluster.Directory {
if (arg0 == null) {
throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument == null");
}
if (arg0.getUrl() == null) {
throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument getUrl() == null");
}
com.alibaba.dubbo.common.URL url = arg0.getUrl();
String extName = url.getParameter("cluster", "failover");
if (extName == null) {
throw new IllegalStateException(
"Fail to get extension(com.alibaba.dubbo.rpc.cluster.Cluster) name from url(" +
url.toString() + ") use keys([cluster])");
}
com.alibaba.dubbo.rpc.cluster.Cluster extension = (com.alibaba.dubbo.rpc.cluster.Cluster) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.cluster.Cluster.class)
.getExtension(extName);
return extension.join(arg0);
}
}

复制代码

doRefer

颠末上面的操纵，如今来到了doRefer函数操纵，其效果返回的就是invoker对象

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// cluster 集群，目前可能为FailoverCluster，如果在有分组的情况下则是MergeableCluster
// registry 注册中心信息
// type 就是上面说的
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
directory.setRegistry(registry);
directory.setProtocol(protocol);
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
&& url.getParameter(Constants.REGISTER_KEY, true)) {
// 接口名字有意义而且是注册协议
// 把当前url信息当做消费者节点注册到注册中心中区
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对象
return cluster.join(directory);
}

复制代码

在上面一笔带过了怎样注册、订阅、天生invoke的，接下来依次拆分各个细节
注册到注册中央

registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY, Constants.CHECK_KEY, String.valueOf(false)));

复制代码

registry是ZookeeperRegistry对象
subscribeUrl是consumer://192.168.10.123/com.jwfy.dubbo.product.ProductService?application=dubbo-consume&default.check=false&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=1196&side=consumer&timestamp=1526204222984

表现是一个斲丧者

进入到FailbackRegistry类

public void register(URL url) {
super.register(url);
// 会把当前的url添加到registered集合中，表示该url注册了
failedRegistered.remove(url);
failedUnregistered.remove(url);
// 既然都要注册了，肯定从失败的集合和取消注册的集合中移除掉
try {
// 向服务器端发送注册请求，也是真正的开始注册操作
doRegister(url);
} catch (Exception e) {
Throwable t = e;
// 如果开启了启动时检测，则直接抛出异常
boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
&& url.getParameter(Constants.CHECK_KEY, true)
&& ! Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
boolean skipFailback = t instanceof SkipFailbackWrapperException;
if (check || skipFailback) {
if(skipFailback) {
t = t.getCause();
}
throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
} else {
logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
// 将失败的注册请求记录到失败列表，定时重试
failedRegistered.add(url);
}
}
// 关于failedRegistered的重试操作，在构造函数中有
// 开启定时任务运行的操作，默认时间是5秒执行一次
this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
public void run() {
// 检测并连接注册中心
try {
retry();
// 里面肯定有遍历failedRegistered集合的doRegister操作（事实上确实有）
} catch (Throwable t) { // 防御性容错
// 个人觉得这个代码写的很好，因为突出一个点，什么时候抛出异常，什么时候处理异常
// 这点问题上，自己曾经踩了很多坑，也发现很多人也有这个毛病
logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
}
}
}, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);

复制代码

在doRegister操纵中，是利用zk的API存储如下的path
/dubbo-jwfy/com.jwfy.dubbo.product.ProductService/consumers/consumer%3A%2F%2F192.168.10.123%2Fcom.jwfy.dubbo.product.ProductService%3Fapplication%3Ddubbo-consume%26category%3Dconsumers%26check%3Dfalse%26default.check%3Dfalse%26dubbo%3D2.5.3%26interface%3Dcom.jwfy.dubbo.product.ProductService%26methods%3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D1196%26side%3Dconsumer%26timestamp%3D1526204222984
如下图，在调用doRegister前后zk注册中央节点的情况，很显着已经注册乐成

订阅服务

服务订阅说普通些就是获取zk中须要的节点信息，本例中是获取生产者的毗连信息

directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
// 这个url 添加了生产者、配置、路由三个节点的配置参数

复制代码

url是consumer://192.168.10.123/com.jwfy.dubbo.product.ProductService?application=dubbo-consume&category=providers,configurators,routers&default.check=false&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=1196&side=consumer&timestamp=1526204222984
directory 是RegistryDirectory，此中参数registry就是上面的注册中央ZookeeperRegistry的设置

FailbackRegistry 类

public void subscribe(URL url, NotifyListener listener) {
super.subscribe(url, listener);
// 会设置registry的subscribed信息，其中subscribed是一个Map<Url,Set<NotifyListener>>的容器
removeFailedSubscribed(url, listener);
try {
// 向服务器端发送订阅请求，真正干活的来了
doSubscribe(url, listener);
} catch (Exception e) {
Throwable t = e;
List<URL> urls = getCacheUrls(url);
if (urls != null && urls.size() > 0) {
notify(url, listener, urls);
logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
} else {
// 如果开启了启动时检测，则直接抛出异常
boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
&& url.getParameter(Constants.CHECK_KEY, true);
boolean skipFailback = t instanceof SkipFailbackWrapperException;
if (check || skipFailback) {
if(skipFailback) {
t = t.getCause();
}
throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t);
} else {
logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
}
// 将失败的订阅请求记录到失败列表，定时重试
// 和上面注册的套路一致
addFailedSubscribed(url, listener);
}
}

复制代码

来到了doSubscribe方法,在这里我们将会相识到怎样从注册中央获取到生产者的毗连信息的

protected void doSubscribe(final URL url, final NotifyListener listener) {
try {
if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
// Constants.ANY_VALUE是*
// url.getServiceInterface 是 com.jwfy.dubbo.product.ProductService
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) {
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);
List<String> services = zkClient.addChildListener(root, zkListener);
if (services != null && services.size() > 0) {
anyServices.addAll(services);
for (String service : services) {
subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
Constants.CHECK_KEY, String.valueOf(false)), listener);
}
}
} else {
List<URL> urls = new ArrayList<URL>();
for (String path : toCategoriesPath(url)) {
// 这个就是上面说的三个目录节点，生产者，配置，路由
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
// zkListeners 是 Map<URL, Map<NotifyListener, ChildListener>>
// 管理url与其对应的监听者和zk监听者的映射关系
if (listeners == null) {
zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
listeners = zkListeners.get(url);
}
ChildListener zkListener = listeners.get(listener);
// 从集合中获取到zk的监听者
if (zkListener == null) {
// 如果没有，则需要手动创建一个新的，其中包含了自定义实现的更新子节点的函数操作
// 里面会调用notify订阅方法（这个方法很重要，由zkClient主动调用childChanged方法）
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);
// 创建永久节点，此时的path可能为/dubbo-jwfy/com.jwfy.dubbo.product.ProductService/providers
List<String> children = zkClient.addChildListener(path, zkListener);
// 重点来了。。。。。这里将会获取到对应生产者的连接信息，具体看下面代码
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
notify(url, listener, urls);
}
} catch (Throwable e) {
throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}

复制代码

AbstractZookeeperClient 类

public List<String> addChildListener(String path, final ChildListener listener) {
ConcurrentMap<ChildListener, TargetChildListener> listeners = childListeners.get(path);
// 从zkClient本身的childListeners获取path的监听者信息
if (listeners == null) {
childListeners.putIfAbsent(path, new ConcurrentHashMap<ChildListener, TargetChildListener>());
listeners = childListeners.get(path);
}
TargetChildListener targetListener = listeners.get(listener);
// 再获取目标节点信息（这个就包含了生产者的信息了）
if (targetListener == null) {
listeners.putIfAbsent(listener, createTargetChildListener(path, listener));
targetListener = listeners.get(listener);
}
return addTargetChildListener(path, targetListener);
}
public List<String> addTargetChildListener(String path, final IZkChildListener listener) {
return client.subscribeChildChanges(path, listener);
// 这一步就会深入到zkClientjar包内进行最后的_zk.getChildren操作
// 返回的List<String> 就是对应的path路径的内容
// 此时path 是 /dubbo-jwfy/com.jwfy.dubbo.product.ProductService/providers
// 返回的内容是 dubbo%3A%2F%2F192.168.10.123%3A20880%2Fcom.jwfy.dubbo.product.ProductService
// %3Fanyhost%3Dtrue%26application%3Ddubbo-demo%26default.loadbalance%3Drandom%26
// dubbo%3D2.5.3%26interface%3Dcom.jwfy.dubbo.product.ProductService%26
// methods%3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D1081%26side%3Dprovider%26
// timestamp%3D1526198287386%26token%3Dfdfdf
}

复制代码

如今完成了和注册中央的操纵了，通过path顺遂拿到生产者的信息，假如过细观察上述的参数信息，会发现pid是1081，再看看jps表现的历程号,如下图，恰好分析获取到的生产者信息是对的

接着来到toUrlsWithEmpty函数，假如有过细观察这个方法会发现，参数列表都改成了(URL consumer, String path, List providers)这已经很明白的告诉我们，第一个参数是斲丧者的url，第二个是当前zk的path信息，第三个是获取到的生产者列表信息（为啥是列表呢？由于生产者可以是多个，而且存在多个的情况下，后续均衡负责还须要选择一个可用的生产者举行网络信息交互操纵）
固然toUrlsWithEmpty函数紧张是举行生产者和斲丧者的url信息对比操纵，假如没有符合的url则添加一个empty协议的url信息（后期就是通过这个empty判定是否存在有效的生产者，一样平常开辟中的无效优劣名单的错误就产生在这里）
革新invoker

紧接着来到了notify方法，如下图的详细各个参数详细值

紧接着来到了AbstractRegistry类

protected void notify(URL url, NotifyListener listener, List<URL> urls) {
// url 是注册到注册到注册中心的url
// listener是监听器，其实就是RegistryDirectory
// urls 是从注册中心获取到到的服务提供方的url集合
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);
}
Map<String, List<URL>> result = new HashMap<String, List<URL>>();
for (URL u : urls) {
if (UrlUtils.isMatch(url, u)) {
// isMatch(URL consumerUrl, URL providerUrl) 函数的参数
// 已经非常清楚的说明了url是服务调用方的url，u是服务提供方的url
// 也和我们上面的描述是一致的
// 对两个url的接口、类目、enable、分组、版本号、classifier等内容进行匹配
// 如果匹配合适，就认为为true
// 没有针对协议进行匹配操作
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);
}
}
// 按照类目进行分组成为一个map
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);
// 替换notified.get(url)的map信息
saveProperties(url);
// 初次看，这个url就是上面的服务调用方的url信息，没必要每次都保存吧
// 进入到这个函数可以发现，他会每次又调用notified.get(url)去获取最新的map数据，默认为异步保存数据
listener.notify(categoryList);
// 接着来到了RegistryDirectory的notify方法
}
}

复制代码

RegistryDirectory 类

public synchronized void notify(List<URL> 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
// 合并override参数
this.overrideDirectoryUrl = directoryUrl;
if (localConfigurators != null && localConfigurators.size() > 0) {
for (Configurator configurator : localConfigurators) {
this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
}
}
// invokerUrls只能是服务提供方目录去刷新已存的invoke
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())) {
// empty协议，而且只有1个
// 需要注意到这个方法只有服务提供方的url才可以被调用
// 服务提供方只包含了一个empty协议的无效url，设置forbidden=true
// 这个true就是后面dubbo中经常出现的黑白名单错误情况
this.forbidden = true; // 禁止访问
this.methodInvokerMap = null; // 置空列表
destroyAllInvokers(); // 关闭所有Invoker
} else {
this.forbidden = false; // 允许访问
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);//缓存invokerUrls列表，便于交叉对比
}
if (invokerUrls.size() ==0 ){
return;
}
// 后面的操作就是去刷新现存的invoker列表
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
// 这个invoker是InvokerDelegete类，是包装了一层的InvokerWrapper
Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 换方法名映射Invoker列表
// state change
//如果计算错误，则不进行处理.
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); // 关闭未使用的Invoker
}catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}

复制代码

这个函数详细的操纵包罗了

更新ZookeeperRegistry中的notify参数信息（把生产者、设置、路由等url信息存入此中）
生存节点信息文档到dubbo的cache文件中
革新天生invoke的数据信息

到此整个的订阅服务的操纵就完成了
天生Invoker

如今就剩下最关键的一句话cluster.join(directory)，会层层包装，末了形成的invoker如图所示

在directory中包罗了全部的注册信息，在反面的真正的函数调用实在也是通过invoker.invoker去调用实行
InvokerInvocationHandler 入口

被包装的类通过动态署理反射时，内嵌入的InvocationHandler类，详细方法调用则会进入到invoke方法中

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
// 获取方法的名称以及方法的参数信息
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
// 此时invoker是MockClusterInvoker
// 还拼接生成了一个RpcInvocation
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}

复制代码

来到了MockClusterInvoker类，此时须要留意到MockClusterInvoker类的invoke是FailoverClusterInvoker
FailoverClusterInvoker类可以举行重试操纵，假如有印象的可以知道在一个reference的xml设置中，可以加上重试次数retries属性字段的值，默认是3次，假如设置了小于0的数字，则为1次，重试次数0位的意思就是只举行一次操纵
MockClusterInvoker mock入口

public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
// 注册中心，服务提供方、服务调用方的信息都存储在directory中，后期均衡负责也是处理这里面的数据
String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
// 就是查看 `methodName.mock`或者`mock`的属性值，默认是“false”
if (value.length() == 0 || value.equalsIgnoreCase("false")){
// 不需要走Mock测试，进入到FailoverClusterInvoker中
result = this.invoker.invoke(invocation);
} else if (value.startsWith("force")) {
if (logger.isWarnEnabled()) {
logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
}
//force:direct mock
result = doMockInvoke(invocation, null);
} else {
//fail-mock
try {
result = this.invoker.invoke(invocation);
}catch (RpcException e) {
if (e.isBiz()) {
throw e;
} else {
if (logger.isWarnEnabled()) {
logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
}
result = doMockInvoke(invocation, e);
}
}
}
return result;
}

复制代码

AbstractClusterInvoker 负载均衡

进入到FailoverClusterInvoker类之前先辈入到AbstractClusterInvoker类中

public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
LoadBalance loadbalance;
List<Invoker<T>> invokers = list(invocation);
// 筛选出合适的invokers列表，基于方法和路由信息
// 其中路由信息则是通过MockInvokersSelector类处理获取到invocation中attachments保存的mock信息去筛选合适的invoker，所以重点是筛选
// 调试发现，一般情况下在这里面attachments字段并没有数据
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
// 创建合适的均衡负责类loanbalance信息，一般情况是RandomLoadBalance类
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 确保幂等，如果是异步则需要往attachment参数中添加自增ID（这个自增ID是AtomicLong类，线程安全）
// 这里就有往invocation的attachment填充数据的操作
return doInvoke(invocation, invokers, loadbalance);
// 现在进入到FailoverClusterInvoker类中了
}

复制代码

FailoverClusterInvoker 重试机制

上面已经说了，这个类的紧张作用是重试操纵
[code]public Result doInvoke(Invocation invocation, final List invokers, LoadBalance loadbalance) throws RpcException { // 第一个是实行的参数信息，包罗了函数名等信息 // 第二个是被调用实行的参数，包罗了服务提供方的IP

ORT信息 // 第三个是均衡负责，在选择调用的服务方时，会根据该对象选择一个符合的服务方 List copyinvokers = invokers; checkInvokers(copyinvokers, invocation); // 检测invokers是否存在，假如不存在则提示没有可用的服务提供方被利用，请查抄服务提供方是否被注册 int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1; // 获取重试的次数，假如设置的值

dubbo服务调用源码

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

浏览过的版块

南七星之家