深入理解dubbo之服务引用

在之前说了一下dubbo的服务发布过程，其实严格意义上只说了一半吧，只把dubbo如何经过ProxyFactory的代理成一个Invoker，等待客户端调用的过程讲了一遍，而重要的Protocol.export方法略过去了，今天我将连带dubbo的comsumer客户端服务引用和Protocol机制来讲一讲。

dubbo服务引用

和上一篇文章一样，先来个demo

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd"> 
<dubbo:application  name="ifenqu-web"  />
<dubbo:registry protocol="zookeeper" address="${dubbo.address}"  />
<dubbo:protocol name="dubbo" port="${dubbo.port}"/>
<dubbo:consumer timeout="60000" check="false"/>
<dubbo:service interface="com.xxx.xxx.xxxService" ref="xxxService" timeout="5000"/>

</beans>

在上篇文章我们已经说了，对于service暴露方也就是provider方有对应的ServiceConfig，相应的Reference引用方也就是Consumer有对应的ReferenceConfig。ReferenceConfig中定义了每一个接口参数定义，这只是部分，还有一大堆参数在父类里就不列出来了。

// 接口类型
private String               interfaceName;

private Class<?>             interfaceClass;

// 客户端类型
private String               client;

// 点对点直连服务提供地址
private String               url;

// 方法配置
private List<MethodConfig>   methods;//接口所有的方法配置

// 缺省配置
private ConsumerConfig       consumer;//该参数对应的就是<dubbo:consumer timeout="60000" check="false"/>

private String               protocol;//如果为空默认dubbo

参数设值分两步，第一步是对象创建的时候，第二步是调用了get方法后执行init()方法。这个get方法就是服务引用的入口：

public synchronized T get() {
if (destroyed){
throw new IllegalStateException("Already destroyed!");
}
//服务实例已经存在就直接返回，没有就进行初始化
if (ref == null) {
init();
}
return ref;
}

init()方法主要分为两个步骤，第一步：收集上下文，第二步：根据上下文创建服务实例

ref = createProxy(map);

进入createProxy方法中去看看

private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp", "localhost", 0, map);
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
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);
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.");
}
}

if (urls.size() == 1) {
invoker = refprotocol.refer(interfaceClass, urls.get(0));
} else {
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; // 用了最后一个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));
}
}
}

//忽略非核心代码
// 创建服务代理
return (T) proxyFactory.getProxy(invoker);
}

该方法中主要逻辑就是先判断需要引用的类型，是本地服务暴露还是直连远程服务还是集群远程服务。如果暴露的服务本地就有直接url就是localhost，而对于集群还涉及到了loadbanlance。无论是什么类型的服务核心都是

refprotocol.refer(interfaceClass, urls.get(0))

，这个方法的返回值就是上篇说的Invoker对象，回忆一下，Invoker对象中封装了接口信息和invoke方法，只要客户端拿到了这个Invoker就可以执行invoke进而通过远程通信触发服务端的service返回执行结果。让我们的视线再回到refprotocol上：

@SPI("dubbo")
public interface Protocol {
int getDefaultPort();

@Adaptive
<T> Exporter<T> export(Invoker<T> var1) throws RpcException;

@Adaptive
<T> Invoker<T> refer(Class<T> var1, URL var2) throws RpcException;

void destroy();
}

这个Protocol接口定义了三个方法，export、refer、destory。分别是服务暴露和服务引用和销毁方法。经过上一篇的讲述，我们已经知道了dubbo支持dubbo、http、thrift等多种协议。那么dubbo是如何做到多个版本协议可以切换自如呢？方法就在SPI上。

SPI

SPI（Service Provider Interface）本来是针对不同厂商或插件的一个规范，提供扩展的时候可以对同一个功能用不同的实现。Java SPI的基本思想可以用设计模式六大原则之开闭原则解释，也就是说要对接口开放，对修改关闭。基于这个原则我们就可以对不同的实现完成可拔插的效果。多种不用实现想用哪种用哪种，只要简单修改配置。

Java SPI的具体约定为:当服务的提供者，提供了服务接口的一种实现之后，在jar包的META-INF/services/目录里同时创建一个以服务接口命名的文件。该文件里就是实现该服务接口的具体实现类。而当外部程序装配这个模块的时候，就能通过该jar包META-INF/services/里的配置文件找到具体的实现类名，并装载实例化，完成模块的注入。 基于这样一个约定就能很好的找到服务接口的实现类，而不需要再代码里制定。jdk提供服务实现查找的一个工具类：java.util.ServiceLoader

知道了SPI的定义现在就更好了解Protocol的实现原理了，在ReferenceConfig中获取具体的Protocol是这一行代码

private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

我们可以根据SPI的定义来看看它的实现过程。进入ExtensionLoader类，在这个类中我们可以看到以下几个熟悉的全局常量

private static final String SERVICES_DIRECTORY = "META-INF/services/";
private static final String DUBBO_DIRECTORY = "META-INF/dubbo/";
private static final String DUBBO_INTERNAL_DIRECTORY = "META-INF/dubbo/internal/";
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap();

debug断点跟进去就会看到这个EXTENSION_LOADERS里面已经装满了多个ExtensionLoaders，尽到dubbo的jar包的

META-INF/dubbo/internal/

路径时就全明白了，这里面就是这些ExtensionLoaders。就拿我们要说的Protocol来说，

ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension()

这行代码说的就是传入Protocol.class类型，就能在EXTENSION_LOADERS中找到

com.alibaba.dubbo.rpc.Protocol

这个类，也就是

META-INF/dubbo/internal/

路径下的定义的文件

com.alibaba.dubbo.rpc.Protocol

。进入这个文件我们能看到所有Protocol实现类定义:

registry=com.alibaba.dubbo.registry.integration.RegistryProtocol
dubbo=com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol
filter=com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper
listener=com.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper
mock=com.alibaba.dubbo.rpc.support.MockProtocol
injvm=com.alibaba.dubbo.rpc.protocol.injvm.InjvmProtocol
rmi=com.alibaba.dubbo.rpc.protocol.rmi.RmiProtocol
hessian=com.alibaba.dubbo.rpc.protocol.hessian.HessianProtocol
com.alibaba.dubbo.rpc.protocol.http.HttpProtocol
com.alibaba.dubbo.rpc.protocol.webservice.WebServiceProtocol
thrift=com.alibaba.dubbo.rpc.protocol.thrift.ThriftProtocol
memcached=com.alibaba.dubbo.rpc.protocol.memcached.MemcachedProtocol

redis=com.alibaba.dubbo.rpc.protocol.redis.RedisProtocol

如果你选择dubbo协议，Protocol的接口实现类就会使用

com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol

我这里使用的是系统默认的dubbo协议，所以我们上面ReferenceConfig方法里面调用的

invoker = refprotocol.refer(interfaceClass, urls.get(0));

就是DubboProtocol的Refer方法：

public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
DubboInvoker<T> invoker = new DubboInvoker(serviceType, url, this.getClients(url), this.invokers);
this.invokers.add(invoker);
return invoker;
}

这个方法没什么好说的，就是Invoker的实例化而已，不过这个invoker已经有所有调用服务端的服务的必要参数。只需要通过invoker作为参数用ProxyFactory进行动态代理来拿到代理类就行了。

远程调用

真正在方法调用时才会触发invoker的doInvoke方法，让我们看看这个doInvoke方法：

protected Result doInvoke(Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation)invocation;
String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment("path", this.getUrl().getPath());
inv.setAttachment("version", this.version);
//通信客户端，可以与socket的客户端类比
ExchangeClient currentClient;
if(this.clients.length == 1) {
currentClient = this.clients[0];
} else {
currentClient = this.clients[this.index.getAndIncrement() % this.clients.length];
}

try {
boolean isAsync = RpcUtils.isAsync(this.getUrl(), invocation);
boolean isOneway = RpcUtils.isOneway(this.getUrl(), invocation);
int timeout = this.getUrl().getMethodParameter(methodName, "timeout", 1000);
if(isOneway) {
boolean isSent = this.getUrl().getMethodParameter(methodName, "sent", false);
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture((Future)null);
return new RpcResult();
} else if(isAsync) {
ResponseFuture future = currentClient.request(inv, timeout);
RpcContext.getContext().setFuture(new FutureAdapter(future));
return new RpcResult();
} else {
RpcContext.getContext().setFuture((Future)null);
return (Result)currentClient.request(inv, timeout).get();
}
} catch (TimeoutException var9) {
throw new RpcException(2, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + this.getUrl() + ", cause: " + var9.getMessage(), var9);
} catch (RemotingException var10) {
throw new RpcException(1, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + this.getUrl() + ", cause: " + var10.getMessage(), var10);
}
}

这个方法也很简单，首先拿到远程调用的参数，比如方法名，调用路径，暴露服务的版本号（版本号不同无法调通），拿到了ExchangeClient后就开始了请求，这个请求分同步和异步，都是看你配置来的，如果是同步的就一直阻塞直到timeout或者结果返回，如果是异步那么直接返回一个ResponseFuture，等执行成功后提供回调。到这，从consumer的方法引用到方法执行都说完了。讲得比较精炼，把很多东西都给省略了，其实dubbo真的特别复杂，但是对于我个人来说只要了解原理就已经达到我的目的了，所以到这就可以了。其实还有一部分比较重要，那就是网络通信部分，dubbo用的是netty。等有空了可以去膜拜膜拜。