RPC调用在客户端（Consumer）触发，基配置文件中会有如下的定义：
<dubbo:reference id="xxxService" interface="xxx.xxx.Service" />
这一行定义会为服务接口xxx.xxx.Service在本地生成一个远程代理，在Dubbo中这个代理用com.alibaba.dubbo.common.bytecode.proxy0的实现来表示，另外，由于这个代理存在于本地，因此就可以像本地bean一样调用该服务，具体的通信过程由代理负责。这个代理实例中仅仅包含一个handler对象（InvokerInvocationHandler类的实例），handler中则包含了RPC调用中非常核心的一个接口Invoker<T>的实现。
Invoker<T>接口的核心方法是invoker(Invocation invocation)，方法的参数Invocation是一个调用过程的抽象，也是Dubbo框架的核心接口，该接口中包含如何获取调用方法的名称、参数类型列表、参数列表以及绑定的数据。如果是RPC调用时，Invocation的具体实现就是RPCInvocation，该方法会抛出RpcException。
代理中的handler实例中包含的Invoker<T>接口实现者是MockClusterInvoker，其中MockClusterInvoker仅仅是一个Invoker的包装，其只是用于实现Dubbo框架中的mock功能。
public Result invoke(Invocation invocation) throws RpcException { Result result = null; //用于获取该服务是否提供mock功能，如果提供，则url中会包含mock关键字 String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim(); if (value.length() == 0 || value.equalsIgnoreCase("false")){ //no mock，直接调用 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 ，用于强制mock的情况，不执行远程调用 result = doMockInvoke(invocation, null); } else { //fail-mock，处理失败后mock的情况，即出现异常时执行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; }
MockClusterInvoker只是包装了一个Invoker的真实实现，其中真实的Invoker实现包含了集群容错的功能，Invoker接口的实现者包括：
AvailbaleClusterInvokerBroadcastClusterInvokerFailbackClusterInvokerFailfastClusterInvokerFailoverClusterInvokerFailsafeClusterInvokerForkingClusterInvoker
默认是FailoverClusterInvoker的实现，即失败重试。进入FailoverCluserInvoker的实现中，可以发现其核心方法invoker(Invocation inocation)的继承自AbstractClusterInvoker，AbstractClusterInvoker的invoker方法：
public Result invoke(final Invocation invocation) throws RpcException { checkWheatherDestoried(); LoadBalance loadbalance; List<Invoker<T>> invokers = list(invocation); if (invokers != null && invokers.size() > 0) { URL invokerUrl = invokers.get(0).getUrl(); loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokerUrl.getMethodParameter(invocation.getMethodName(), (invokerUrl.hasParameter(Constants.DEFAULT_LOADBALANCE_KEY)?Constants.DEFAULT_LOADBALANCE_KEY:Constants.LOADBALANCE_KEY), Constants.DEFAULT_LOADBALANCE)); } else { loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE); } RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation); return doInvoke(invocation, invokers, loadbalance); }调用过程中首先通过方法list(invocation)获取Invoker的列表List<Invoker<T>>，这个列表可以对应到服务提供者的列表。其中list(invocation)方法的实现代码如下：
protected List<Invoker<T>> list(Invocation invocation) throws RpcException { List<Invoker<T>> invokers = directory.list(invocation); return invokers; }
这里用到另一个比较关键的接口Directory，它是Dubbo框架中用于封装服务提供者列表的一个数据结构。它的list方法根据Inovcation对象（一次远程调用过程），返回一个调用者的列表，其中Directory接口实例是RegistryDirectory类的对象，RegistryDirectory的list方法继承自AbstractDirectory。
public List<Invoker<T>> list(Invocation invocation) throws RpcException { if (destroyed){ throw new RpcException("Directory already destroyed .url: "+ getUrl()); } List<Invoker<T>> invokers = doList(invocation); List<Router> localRouters = this.routers; // local reference if (localRouters != null && localRouters.size() > 0) { for (Router router: localRouters){ try { if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) { invokers = router.route(invokers, getConsumerUrl(), invocation); } } catch (Throwable t) { logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t); } } } return invokers; }AbstractDirectory类的list方法首先会获取当前Invocation的调用列表，由抽象方法doList(invocation)实现。接下来就是对获取到的当前Invocation的调用列表List<Invoker<T>>进行路由处理，每个Consumer会在本地缓存（或者从注册中心获取 ）路由集合，然后判断集合中的每一个路由规则是否可以对当前调用进行过滤，如果路由规则符合当前调用，就对调用列表List<Invoker<T>>进行筛选。
Router接口的实现主要有两个，一个是ConditionRouter，另一个是ScriptRouter。这里说一下条件路由
RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension(); Registry registry = registryFactory.getRegistry(URL.valueOf("zookeeper://10.20.153.10:2181")); registry.register(URL.valueOf("condition://0.0.0.0/com.foo.BarService?category=routers&dynamic=false&rule=" + URL.encode("http://10.20.160.198/wiki/display/dubbo/host = 10.20.153.10 => host = 10.20.153.11") + "));其中：
conditon:// 表示路由规则的类型，支持条件路由和脚本路由，可扩展。0.0.0.0 表示对所有IP地址生效。com.foo.BarService 表示只对指定服务生效。category=routers 表示该数据为动态配置类型。dynamic=false 表示该数据为持久数据，当注册方退出时，数据依然保存在注册中心。enabled=true 覆盖规则是否生效，可不填。force=false 当路由结果为空时，是否强制执行，如果不强制执行，路由结果为空的路由规则将自动失效，可不填。runtime=false 是否在每次调用时执行路由规则，否则只在提供者地址列表变更时预先执行并缓存结果，调用时真接从缓存中获取路由结果。如果用了参数路由，必须设为true，可不填。priority=1 路由规则的优先级，用于排序，优先级越大越靠前执行，可不填。rule=URL.encode("host=10.20.153.10=>host=10.20.153.11") 表示路由规则的内容。规则说明：
=>之前的为消费者匹配条件，所有参数和消费者的URL进行对比，当消费者满足匹配条件时，对该消费者执行后面的过滤规则。=>之后的为提供者地址列表的过滤条件，所有参数的提供者的URL进行对比。表达式：
参数支持：（1）服务调用信息，如method,argument等（暂不支持参数路由）
（2）URL本身的字段，如protocl, host, port等。
（3）以及URL上的所有参数，如application, organization等。
条件支持：（1）=
（2）！=
值支持：（1）逗号分隔多个值。
（2）星号表示通配。
（3）$开头，表示引用消费者参数。
示例：
排除预发机：=> host != 172.22.3.91
白名单host != 20.20.153.10
黑名单host=20.20.153.10
只暴露一部分机器，防止整个集群挂掉=>host=172.22.3.1*
为重要应用提供额外的机器application!=kylin => host!=172.22.3.95
读写分离method=find*,list*,get*,is*=>host=172.22.3.97
method!=find*,list*,get*,is*=>172.22.3.97
隔离不同机房网段
host!=172.22.3.* =>host!=172.22.3.*
提供者和消费者部署在同集群内，本机只访问本机的服务=> host =$host
ConditionRouter的route方法：
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL consumerUrl, Invocation invocation) throws RpcException { if (invokers == null || invokers.size() == 0) { return invokers; } try { // 白名单过滤 if(matchWhen(consumerUrl, Constants.WHITE_LIST)){ invokers = new ArrayList<Invoker<T>>(); throw new RpcException("The current consumer not in the service whitelist. consumer: " + NetUtils.getLocalHost() + ", service: " + consumerUrl.getServiceKey() + "."); } // 黑名单过滤 if(matchWhen(consumerUrl, Constants.BLACK_LIST)){ invokers = new ArrayList<Invoker<T>>(); throw new RpcException("The current consumer in the service blacklist. consumer: " + NetUtils.getLocalHost() + ", service: " + consumerUrl.getServiceKey() + "."); } List<Invoker<T>> result = new ArrayList<Invoker<T>>(); for (Invoker<T> invoker : invokers) { if (matchThen(invoker.getUrl(), consumerUrl)) { result.add(invoker); } } if (result.size() > 0) { return result; } else if (force) { logger.warn("The route result is empty and force execute. consumer: " + NetUtils.getLocalHost() + ", service: " + consumerUrl.getServiceKey() + ", router: " + consumerUrl.getParameterAndDecoded(Constants.RULE_KEY)); return result; } } catch (Throwable t) { logger.error("Failed to execute condition router rule: " + getUrl() + ", invokers: " + invokers + ", cause: " + t.getMessage(), t); RpcContext.getContext().set(Constants.ROUTER_KEY, t.getMessage()); } return invokers; }
消费者匹配规则（=>之前的条件）用whencondition表示，提供者地址列表滤规则用thencondition表示，如果匹配条件为空，表示对所有消费者生效，如果过滤条件为空，表示禁止访问。
ConditionRouter判断whencondition是否匹配当前调用时仅仅根据consumer的url来判断，虽然调用实现Invocation也作为参数传入，但是并没有参与whencondition的匹配过程。
AbstractClusterInvoker的invoker方法完成了对Invocation的调用者路由之后，会获取负载均衡实例LoadBalance，然后根据当前Invocation实例、调用者列表和负载均衡实现执行AbstractClusterInvoker.doInvoker抽象方法。
FailoverClusterInvoer的doInvoker方法：
@SuppressWarnings({ "unchecked", "rawtypes" }) public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { List<Invoker<T>> copyinvokers = invokers; checkInvokers(copyinvokers, invocation); int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1; if (len <= 0) { len = 1; } // retry loop. RpcException le = null; // last exception. List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers. Set<String> providers = new HashSet<String>(len); for (int i = 0; i < len; i++) { //重试时，进行重新选择，避免重试时invoker列表已发生变化. //注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变 if (i > 0) { checkWheatherDestoried(); copyinvokers = list(invocation); //重新检查一下 checkInvokers(copyinvokers, invocation); } Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked); invoked.add(invoker); RpcContext.getContext().setInvokers((List)invoked); try { Result result = invoker.invoke(invocation); if (le != null && logger.isWarnEnabled()) { logger.warn("Although retry the method " + invocation.getMethodName() + " in the service " + getInterface().getName() + " was successful by the provider " + invoker.getUrl().getAddress() + ", but there have been failed providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the hsf version " + Version.getVersion() + ". Last error is: " + le.getMessage(), le); } URL invokerUrl = invoker.getUrl(); if(invokerUrl != null){ logger.debug("The invoker for " + invokerUrl.getServiceInterface() + " was successful by the provider " + invokerUrl.getAddress()); } return result; } catch (RpcException e) { if (e.isBiz()) { // biz exception. throw e; } le = e; } catch (Throwable e) { le = new RpcException(e.getMessage(), e); } finally { providers.add(invoker.getUrl().getAddress()); } } throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method " + invocation.getMethodName() + " in the service " + getInterface().getName() + ". Tried " + len + " times of the providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the hsf version " + Version.getVersion() + ". Last error is: " + (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le); }根据调用者列表进行一次调用，直到调用成功或者达到retries次数的上限，关于负载均衡部分，其逻辑在AbstractClusterInvoker的方法select，该方法会调用LoadBalance接口的select方法。默认情况下的负载均衡实现是RandomLoadBalance，它会按照特定的权重来随机从提供乾列表中选择一个进行调用。
另外，远程调用在实际执行时会添加一个过滤器链，过滤器链的控制逻辑在ProtocolFilterWrapper类的buildInvokerChain()方法中实现，并由Invoker的实现者触发。
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) { Invoker<T> last = invoker; List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); if (filters.size() > 0) { for (int i = filters.size() - 1; i >= 0; i --) { final Filter filter = filters.get(i); final Invoker<T> next = last; last = new Invoker<T>() {//匿名类 public Class<T> getInterface() { return invoker.getInterface(); } public URL getUrl() { return invoker.getUrl(); } public boolean isAvailable() { return invoker.isAvailable(); } public Result invoke(Invocation invocation) throws RpcException { return filter.invoke(next, invocation); } public void destroy() { invoker.destroy(); } @Override public String toString() { return invoker.toString(); } }; } } return last; }在默认情况下过滤器链的调用顺序为:ConsumerContextFilter->MonitorFilter->FutureFilter，ConsumerContextFilter过滤器主要处理消费者调用时携带的上下文信息，并附加到RpcContext中。MonitorFilter主要对调用过程进行监控，其核心代码是collect方法，该方法会将监控的数据通过Monitor进行收集，收集合存入在本地内存，每隔固定的时间上传到监控节点上。Monitor的默认实现是DubboMonitor，DubboMonitor中会定义一个ConcurrentHahsMap，MonitorFilter的collect方法在执行时会将监控数据保存到该ConcurrentHahsMap中，另外，DubboMonitor默认会开启一个基于线程池的定时任务执行器ScheduledExecutorService，并且在构造函数中启动一个周期性执行任务将ConcureentHashMap中的数据发送到节点上，发送逻辑在send()方法中实现。
最后欢迎大家访问我的个人网站：1024s
