深入理解Dubbo-6.服务消费源码分析（上）:https://developer.aliyun.com/article/1414080
RegistryDirectory 订阅
大家还记得在RegistryProtocol中，调用doRefer这个方法吗？
这个doRefer方法中，构建了一个RegistryDirectory，之前我们不明白Directory的含义，所以暂时没有管，但是它的作用很重要，它负责动态维护服务提供者列表。
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { //初始化RegistryDirectory(注册中心的目录) RegistryDirectory<T> directory = new RegistryDirectory(type, url); directory.setRegistry(registry);// 注册中心 directory.setProtocol(this.protocol);// 协议 Map<String, String> parameters = new HashMap(directory.getConsumerUrl().getParameters()); //注册consumer://协议url URL subscribeUrl = new URL("consumer", (String)parameters.remove("register.ip"), 0, type.getName(), parameters); if (directory.isShouldRegister()) { //注册服务消费者的url地址 directory.setRegisteredConsumerUrl(subscribeUrl); registry.register(directory.getRegisteredConsumerUrl()); } directory.buildRouterChain(subscribeUrl); //进行订阅 订阅地址的变化 //subscribe订阅信息消费url、通知监听、配置监听、订阅url //toSubscribeUrl:订阅信息：category、providers、configurators、routers directory.subscribe(toSubscribeUrl(subscribeUrl)); //一个注册中心会存在多个服务提供者，所以在这里需要把多个服务提供者通过cluster.join合并成一个 Invoker<T> invoker = cluster.join(directory); // 只是初始化了一个RegistryDirectory，然后通过 Cluster.join 来返回一个Invoker对象 List<RegistryProtocolListener> listeners = this.findRegistryProtocolListeners(url); if (CollectionUtils.isEmpty(listeners)) { return invoker; } else { //通过RegistryInvokerWrapper进行包装 RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper(directory, cluster, invoker, subscribeUrl); Iterator var11 = listeners.iterator(); while(var11.hasNext()) { RegistryProtocolListener listener = (RegistryProtocolListener)var11.next(); listener.onRefer(this, registryInvokerWrapper); } return registryInvokerWrapper; } }
服务目录
RegistryDirectory是Dubbo中的服务目录，从名字上来看，也比较容易理解，服务目录中存储了一些和服务提供者有关的信息，通过服务目录，服务消费者可获取到服务提供者的信息，比如 ip、端口、服务协议等。通过这些信息，服务消费者就可通过 Netty 等客户端进行远程调用。
所以，再回忆一下之前我们通过Cluster.join去构建Invoker时，传递了一个directory进去，因为Invoker是一个调用器，在发起远程调用时，必然需要从directory中去拿到所有的服务提供者列表，然后再通过负载均衡机制来发起请求。
RegistryDirectory.subscribe
订阅注册中心指定节点的变化，如果发生变化，则通知到RegistryDirectory。Directory其实和服务的注册以及服务的发现有非常大的关联.
public void subscribe(URL url) { // 对消费者的信息进行监听和订阅 this.setConsumerUrl(url); // 当注册中心的事件发生变化的时候，会有一个方法接收事件变化的通知，从而去更改 RegistryDirectory 里面的invoker CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this); this.serviceConfigurationListener = new RegistryDirectory.ReferenceConfigurationListener(this, url); this.registry.subscribe(url, this); }
此时，registry我们知道，它是ListenerRegistryWrapper(ZookeeperRegsistry)对象，我们先不管包装类，直接进入到ZookeeperRegistry这个类中
FailbackRegistry.subscribe
registry.subscribe这个方法，最终会调用FailbackRegistry.subscrbe
其中入参， listener为RegistryDirectory，后续要用到failbackRegistry这个类，从名字就可以看出，它的主要作用就是实现具有故障恢复功能的服务订阅机制，简单来说就是如果在订阅服务注册中心时出现异常，会触发重试机制。
// 通过前面 动态扩展点 ZookeeperRegistryFactory 说明后面会进入到ZookeeperRegistry ，但是没有 subscribe 所以要去看其父类的实现 FailbackRegistry // 下面进入的就是zk的订阅 public void subscribe(URL url, NotifyListener listener) { super.subscribe(url, listener); this.removeFailedSubscribed(url, listener); try { // 进入到 ZookeeperRegistry 里的 doSubscribe方法 // 发送订阅请求到服务端 this.doSubscribe(url, listener); } ..... // 记录失败的注册订阅请求，根据规则进行重试 this.addFailedSubscribed(url, listener); } }
ZookeeperRegistry.doSubscribe
这个方法是订阅，逻辑实现比较多，可以分两段来看，这里的实现把所有Service层发起的订阅以及指定的Service层发起的订阅分开处理。所有Service层类似于监控中心发起的订阅。指定的Service层发起的订阅可以看作是服务消费者的订阅。我们只需要关心指定service层发起的订阅即可
public void doSubscribe(final URL url, final NotifyListener listener) { try { //针对所有service层发起的定于 if ("*".equals(url.getServiceInterface())) { ...... } else { //针对指定的服务地址发起订阅 //toCategoriesPath 会返回三个结果，分别是/providers、/configurations、/routers。 也就是服务启动时需要监听这三个节点下的数据变化 List<URL> urls = new ArrayList<>(); for (String path : toCategoriesPath(url)) { //构建一个listeners集合，其中key是NotifyListener，其中前面我们知道RegistryDirectory实现了这个接口，所以这里的key应该是RegistryDirectory //value表示针对这个RegistryDirectory注册的子节点监听。 ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>()); ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, k, toUrlsWithEmpty(url, parentPath, currentChilds))); zkClient.create(path, false);//创建一个/providers or/configurators、/routers节点 List<String> children = zkClient.addChildListener(path,zkListener); if (children != null) { urls.addAll(toUrlsWithEmpty(url, path, children)); } } //调用notify方法触发监听 notify(url, listener, urls);
FailbackRegistry.notify
调用FailbackRegistry.notify， 对参数进行判断。 然后调用AbstractRegistry.notify方法
protected void notify(URL url, NotifyListener listener, List<URL> urls) { ...... try { // 其实是 ZookeeperRegistry实例 this.doNotify(url, listener, urls); ...... } protected void doNotify(URL url, NotifyListener listener, List<URL> urls) { super.notify(url, listener, urls); // 其调用父类 AbstractRegistry }
AbstractRegistry.notify
protected void notify(URL url, NotifyListener listener, List<URL> urls) { ...... Map<String, List<URL>> result = new HashMap(); Iterator var5 = urls.iterator(); while(var5.hasNext()) { URL u = (URL)var5.next(); if (UrlUtils.isMatch(url, u)) { String category = u.getParameter("category", "providers"); List<URL> categoryList = (List)result.computeIfAbsent(category, (k) -> { return new ArrayList(); }); categoryList.add(u); } } if (result.size() != 0) { Map<String, List<URL>> categoryNotified = (Map)this.notified.computeIfAbsent(url, (ux) -> { return new ConcurrentHashMap(); }); Iterator var11 = result.entrySet().iterator(); while(var11.hasNext()) { Entry<String, List<URL>> entry = (Entry)var11.next(); String category = (String)entry.getKey(); List<URL> categoryList = (List)entry.getValue(); categoryNotified.put(category, categoryList); //通过listener监听categoryList // 前面的监听主要是根据urls进行的，下面的监听就相当于根据 具体的url进行的。 listener.notify(categoryList); // 将注册中心的地址保存在本地缓存 //保存到本地文件中，作为服务地址的缓存信息 this.saveProperties(url); } } } }
上述代码中，我们重点关注 listener.notify ，它会触发一个事件通知，消费端的listener是最开始传递过来的RegistryDirectory，所以这里会触发RegistryDirectory.notify
RegistryDirectory.notify
看到这个代码，大家是不是更进一步理解了，服务地址发生变化和更新时，调用Directory.notify来更新，那么是不是意味着更新后的信息，会同步到Directory中呢？
带着这个思考，我们继续去分析代码。
public synchronized void notify(List<URL> urls) { //对数据进行过滤。 Map<String, List<URL>> categoryUrls = (Map)urls.stream(). filter(Objects::nonNull). filter(this::isValidCategory). filter(this::isNotCompatibleFor26x). collect(Collectors.groupingBy(this::judgeCategory)); //假设当前进来的通知是 providers节点 //判断configurator是否为空，这个节点下的配置，是在dubbo-admin控制台上修改配置时，会先创建一个配置节点到这个路径下，注册中心收到这个变化时会通知服务消费者，服务消费者会根据新的配置重新构建Invoker List<URL> configuratorURLs = (List)categoryUrls.getOrDefault("configurators", Collections.emptyList()); this.configurators = (List)Configurator.toConfigurators(configuratorURLs).orElse(this.configurators); //判断路由规则配置是否为空，如果不为空，同样将路由规则添加到url中。 List<URL> routerURLs = (List)categoryUrls.getOrDefault("routers", Collections.emptyList()); this.toRouters(routerURLs).ifPresent(this::addRouters); // providers 得到服务提供者的地址列表 List<URL> providerURLs = (List)categoryUrls.getOrDefault("providers", Collections.emptyList()); // 前面就是针对url进行改造 ...... // 这个方法就很清楚了，根据监听后修改的url进行刷新重载Invoker this.refreshOverrideAndInvoker(providerURLs); }
refreshOverrideAndInvoker
private void refreshOverrideAndInvoker(List<URL> urls) { //付给DirectoryUrl this.overrideDirectoryUrl(); //刷新invoker this.refreshInvoker(urls); }
- 逐个调用注册中心里面的配置，覆盖原来的url，组成最新的url 放入overrideDirectoryUrl 存储，此时我们没有在dubbo-admin中修改任何配置，所以这里没必要去分析
- 根据 provider urls，重新刷新Invoker
refreshInvoker
从名字可以看到，这里是刷新invoker。怎么理解呢？
当注册中心的服务地址发生变化时，会触发更新。而更新之后并不是直接把url地址存储到内存，而是把url转化为invoker进行存储，这个invoker是作为通信的调用器来构建的领域对象，所以如果地址发生变化，那么需要把老的invoker销毁，然后用心的invoker替代。
// 刷新Invoker private void refreshInvoker(List<URL> invokerUrls) { Assert.notNull(invokerUrls, "invokerUrls should not be null"); // 前面介绍过，如果发现注册中心不可用的时候，依然可以调用，因为能拿到本地缓存文件里面的注册中心协议的地址然后调用 //如果只有一个服务提供者，并且如果是空协议，那么这个时候直接返回进制访问，并且销毁所有的invokers if (((List)invokerUrls).size() == 1 && ((List)invokerUrls).get(0) != null && "empty".equals(((URL)((List)invokerUrls).get(0)).getProtocol())) { this.forbidden = true; this.invokers = Collections.emptyList(); this.routerChain.setInvokers(this.invokers); // 如果服务提供者只有一个并且是空协议，会禁止调用 // 销毁所有的invoker this.destroyAllInvokers(); } else { this.forbidden = false; //获取老的invoker集合 Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; if (invokerUrls == Collections.emptyList()) { invokerUrls = new ArrayList(); } if (((List)invokerUrls).isEmpty() && this.cachedInvokerUrls != null) { ((List)invokerUrls).addAll(this.cachedInvokerUrls); } else { this.cachedInvokerUrls = new HashSet(); this.cachedInvokerUrls.addAll((Collection)invokerUrls); } if (((List)invokerUrls).isEmpty()) { return; } // 构建一个newnivoker // 转换具体协议的invoker信息 //把invokerUrls转化为InvokerMap Map<String, Invoker<T>> newUrlInvokerMap = this.toInvokers((List)invokerUrls); if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) { logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + ((List)invokerUrls).size() + ", invoker.size :0. urls :" + invokerUrls.toString())); return; } List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList(newUrlInvokerMap.values())); this.routerChain.setInvokers(newInvokers); this.invokers = this.multiGroup ? this.toMergeInvokerList(newInvokers) : newInvokers; // 然后赋值到这个里面 //旧的url 是否在新map里面存在，不存在，就是销毁url对应的Invoker this.urlInvokerMap = newUrlInvokerMap; try { this.destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); } catch (Exception var6) { logger.warn("destroyUnusedInvokers error. ", var6); } } }
toInvokers
这个方法中有比较长的判断和处理逻辑，我们只需要关心invoker是什么时候初始化的就行。这里用到了protocol.refer来构建了一个invoker
invoker = new InvokerDelegate<>(protocol.refer(serviceType, url), url,providerUrl);
构建完成之后，会保存在 Map> urlInvokerMap 这个集合中
private Map<String, Invoker<T>> toInvokers(List<URL> urls) { Map<String, Invoker<T>> newUrlInvokerMap = new HashMap(); ............. if (enabled) { invoker = new RegistryDirectory.InvokerDelegate(this.protocol.refer(this.serviceType, url), url, providerUrl); } } catch (Throwable var13) { logger.error("Failed to refer invoker for interface:" + this.serviceType + ",url:(" + url + ")" + var13.getMessage(), var13); } if (invoker != null) { newUrlInvokerMap.put(key, invoker); } } else { newUrlInvokerMap.put(key, invoker); } } } } } } else { return newUrlInvokerMap; } }
总结
简单总结一下，RegistryDirectory.subscribe，其实总的来说就相当于做了两件事
- 定于指定注册中心的以下三个路径的子节点变更事件
/dubbo/org.apache.dubbo.demo.DemoService/providers 、/dubbo/org.apache.dubbo.demo.DemoService/configurators、/dubbo/org.apache. dubbo.demo.DemoService/routers
- 触发时间变更之后，把变更的节点信息转化为Invoker
Protocol.refer
咱们继续往下看，在toInvokers这个方法中，invoker是通过 protocol.refer来构建的。那么我们再来分析一下refer里面做了什么？
protocol.refer(serviceType, url), url, providerUrl)
不过在往下分析之前，我们先来猜想一下它会做什么？根据在这段逻辑所处的位置和它在整个dubbo中的作用，我们应该可以猜测出。
- 这个invoker应该和通信有关系
- 那么这里应该会建立一个网络通信的连接
根据猜想，我们去看一下它的实现，首先protocol，是被依赖注入进来的自适应扩展点Protocol$Adaptive. ，此时传进去的参数，此时url对应的地址应该是dubbo://开头的协议地址，所以最终获得的是通过包装之后的DubboProtocol对象。
QosProtocolWrapper(ProtocolFilterWrapper(ProtocolListenerWrapper(DubboProtocol)))
其中，在Qos这边，会启动Qosserver、在FilterWrapper中，会构建一个过滤器链，其中包含访问日志过滤、访问限制过滤等等，这个最终在调用的时候会通过一条过滤器链路对请求进行处理。
AbstractProtocol.refer
DubboProtocol中没有refer方法，而是调用父类的refer。
@Override public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url)); }
protocolBindingRefer
- 优化序列化
- 构建DubboInvoker
在构建DubboInvoker时，会构建一个ExchangeClient，通过getClients(url)方法，这里基本可以猜到到是服务的通信建立
@Override public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException { optimizeSerialization(url); // create rpc invoker. DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url,getClients(url), invokers); invokers.add(invoker); return invoker; }
getClients
这里面是获得客户端连接的方法
- 判断是否为共享连接，默认是共享同一个连接进行通信
- 是否配置了多个连接通道 connections，默认只有一个共享连接
private ExchangeClient[] getClients(URL url) { boolean useShareConnect = false; int connections = url.getParameter("connections", 0); List<ReferenceCountExchangeClient> shareClients = null; // 如果没有配置的情况下，默认是采用共享连接，否则，就是针对一个服务提供一个连接。 //所谓共享连接，实际上就是 if (connections == 0) { useShareConnect = true; String shareConnectionsStr = url.getParameter("shareconnections", (String)null); connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty("shareconnections", "1") : shareConnectionsStr); //返回共享连接 shareClients = this.getSharedClient(url, connections); } //从共享连接中获得客户端连接进行返回 ExchangeClient[] clients = new ExchangeClient[connections]; for(int i = 0; i < clients.length; ++i) { if (useShareConnect) { clients[i] = (ExchangeClient)shareClients.get(i); } else { //如果不是使用共享连接，则初始化一个新的客户端连接进行返回 clients[i] = this.initClient(url); } } return clients; }
getSharedClient
private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) { String key = url.getAddress(); List<ReferenceCountExchangeClient> clients = (List)this.referenceClientMap.get(key); // //检查当前的key检查连接是否已经创建过并且可用，如果是，则直接返回并且增加连接的个数的统计 if (this.checkClientCanUse(clients)) { this.batchClientRefIncr(clients); return clients; } else { //如果连接已经关闭或者连接没有创建过 this.locks.putIfAbsent(key, new Object()); synchronized(this.locks.get(key)) { clients = (List)this.referenceClientMap.get(key); // 双重检查 if (this.checkClientCanUse(clients)) { this.batchClientRefIncr(clients); return clients; } else { // 连接数必须大于等于1 connectNum = Math.max(connectNum, 1); //如果当前消费者还没有和服务端产生连接，则初始化 if (CollectionUtils.isEmpty(clients)) { clients = this.buildReferenceCountExchangeClientList(url, connectNum); this.referenceClientMap.put(key, clients); } else {//如果clients不为空，则从clients数组中进行遍历 for(int i = 0; i < clients.size(); ++i) { ReferenceCountExchangeClient referenceCountExchangeClient = (ReferenceCountExchangeClient)clients.get(i); // 如果在集合中存在一个连接但是这个连接处于closed状态，则重新构建一个进行替换 if (referenceCountExchangeClient != null && !referenceCountExchangeClient.isClosed()) { //增加个数 referenceCountExchangeClient.incrementAndGetCount(); } else { clients.set(i, this.buildReferenceCountExchangeClient(url)); } } } this.locks.remove(key); return clients; } } } }
buildReferenceCountExchangeClient
开始初始化客户端连接.
也就是说，dubbo消费者在启动的时候，先从注册中心上加载最新的服务提供者地址，然后转化成invoker，但是转化的时候，也会同步去建立一个连接。
而这个连接默认采用的是共享连接，因此就会存在对于同一个服务提供者，假设客户端依赖了多个
@DubboReference，那么这个时候这些服务的引用会使用同一个连接进行通信。(举个例子来说明，假设有两个服务提供者A和B，它们都提供了不同的服务。现在在Dubbo消费者中，我们分别使用了@DubboReference注解来引用A和B的服务。默认情况下，A和B的服务引用会使用同一个连接与对应的服务提供者进行通信。)
这种方式可以有效地减少连接数，提高资源利用率，但也可能会带来一些问题，比如如果某个服务出现了连接问题，可能会影响到其他依赖该服务的引用。因此，开发者在使用Dubbo时，需要根据实际情况来选择合适的连接方式。
private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) { ExchangeClient exchangeClient = initClient(url); return new ReferenceCountExchangeClient(exchangeClient); }
initClient
终于进入到初始化客户端连接的方法了，猜测应该是根据url中配置的参数进行远程通信的构建
private ExchangeClient initClient(URL url) { // 获取客户端通信类型，默认是netty String str = url.getParameter("client", url.getParameter("server", "netty")); //获取编码解码类型，默认是Dubbo自定义类型 url = url.addParameter("codec", "dubbo"); // 开启心跳并设置心跳间隔，默认60s url = url.addParameterIfAbsent("heartbeat", String.valueOf(60000)); // 如果没有找到指定类型的通信扩展点，则抛出异常 if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException("Unsupported client type: " + str + ", supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " ")); } else { try { Object client; // 是否延迟建立连接 if (url.getParameter("lazy", false)) { client = new LazyConnectExchangeClient(url, this.requestHandler); } else {//默认是直接在启动时建立连接 client = Exchangers.connect(url, this.requestHandler); } return (ExchangeClient)client; } catch (RemotingException var5) { throw new RpcException("Fail to create remoting client for service(" + url + "): " + var5.getMessage(), var5); } } }
Exchangers.connect
创建一个客户端连接
public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { if (url == null) { throw new IllegalArgumentException("url == null"); } if (handler == null) { throw new IllegalArgumentException("handler == null"); } url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange"); return getExchanger(url).connect(url, handler); } public static Exchanger getExchanger(URL url) { String type = url.getParameter("exchanger", "header"); return getExchanger(type); } // 动态扩展点 public static Exchanger getExchanger(String type) { return (Exchanger)ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type); }
HeaderExchange.connect
@Override public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true); }
同样Transporters 也是动态扩展点
NettyTransport.connect
使用netty构建了一个客户端连接
@Override public Client connect(URL url, ChannelHandler handler) throws RemotingException { return new NettyClient(url, handler); }
总结
我们讲到了 RegistryProtocol.refer 过程中有一个关键步骤，即在监听到服务提供者url时触发RegistryDirectory.notify() 方法。
RegistryDirectory.notify() 方法调用 refreshInvoker() 方法将服务提供者urls转换为对应的 远程invoker ，最终调用到 DubboProtocol.refer() 方法生成对应的 DubboInvoker 。
DubboInvoker 的构造方法中有一项入参 ExchangeClient[] clients ，即对应本文要讲的网络客户端 Client 。DubboInvoker就是通过调用 client.request() 方法完成网络通信的请求发送和响应接收功能。
Client 的具体生成过程就是通过 DubboProtocol 的 initClient(URL url) 方法创建了一个HeaderExchangeClient 。
ReferenceConfig.createProxy
返回了Invoker之后，再回到createProxy这段代码中，这里最终会调用一个getProxy来返回一个动态代理对象。
并且这里把invoker作为参数传递进去，上面我们知道invoker这个对象的构建过程，实际上封装了一个directory在invoker中。
return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
而这里的proxyFactory是一个自适应扩展点，它会根据url中携带的proxy参数来决定选择哪种动态代理技术来构建动态代理对象，默认是javassist
import org.apache.dubbo.common.extension.ExtensionLoader; public class ProxyFactory$Adaptive implements org.apache.dubbo.rpc.ProxyFactory { public java.lang.Object getProxy(org.apache.dubbo.rpc.Invoker arg0) throws org.apache.dubbo.rpc.RpcException { if (arg0 == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null"); if (arg0.getUrl() == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() ==null"); org.apache.dubbo.common.URL url = arg0.getUrl(); String extName = url.getParameter("proxy", "javassist"); if(extName == null) throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.ProxyFactory) name from url (" + url.toString()+ ") use keys([proxy])"); org.apache.dubbo.rpc.ProxyFactory extension = (org.apache.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(org.apache .dubbo.rpc.ProxyFactory.class).getExtension(extName); return extension.getProxy(arg0); } public java.lang.Object getProxy(org.apache.dubbo.rpc.Invoker arg0, boolean arg1) throws org.apache.dubbo.rpc.RpcException { if (arg0 == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null"); if (arg0.getUrl() == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() ==null"); org.apache.dubbo.common.URL url = arg0.getUrl(); String extName = url.getParameter("proxy", "javassist"); if(extName == null) throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.ProxyFactory) name from url (" + url.toString()+ ") use keys([proxy])"); org.apache.dubbo.rpc.ProxyFactory extension = (org.apache.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(org.apache .dubbo.rpc.ProxyFactory.class).getExtension(extName); return extension.getProxy(arg0, arg1); } public org.apache.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, org.apache.dubbo.common.URL arg2) throws org.apache.dubbo.rpc.RpcException { if (arg2 == null) throw new IllegalArgumentException("url == null"); org.apache.dubbo.common.URL url = arg2; String extName = url.getParameter("proxy", "javassist"); if(extName == null) throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.ProxyFactory) name from url (" + url.toString()+ ") use keys([proxy])"); org.apache.dubbo.rpc.ProxyFactory extension = (org.apache.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(org.apache .dubbo.rpc.ProxyFactory.class).getExtension(extName); return extension.getInvoker(arg0, arg1, arg2); } }
JavassistProxyFactory.getProxy
通过这个方法生成了一个动态代理类，并且对invoker再做了一层处理，InvokerInvocationHandler。意味着后续发起服务调用的时候，会由InvokerInvocationHandler来进行处理。
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) { return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker)); }
proxy.getProxy
在proxy.getProxy这个方法中会生成一个动态代理类，通过debug的形式可以看到动态代理类的原貌在getProxy这个方法位置加一个断点
proxy = (Proxy) pc.newInstance();
然后在debug窗口，找到ccp这个变量 -> mMethods。
从这个sayHello方法可以看出，我们通过@Reference注入的一个对象实例本质上就是一个动态代理类，通过调用这个类中的方法，会触发handler.invoke(), 而这个handler就是InvokerInvocationHandler
以上的步骤都是属于消费者启动的时候所做的动作
