转载

Dubbo源码分析（六）服务引用的具体流程

在前面的章节中，我们已经完成Dubbo服务暴露的流程分析。今天我们一起来看Dubbo怎么引用这些服务的。

关于服务引用，Dubbo有两种方式。一种是基于注册中心进行服务引用，一种是服务直连进行引用。服务直连主要用于测试联调阶段，生产环境不推荐使用。它的配置也比较简单，在消费者端指定服务url即可。

<dubbo:reference id="infoUserService"
    interface="com.viewscenes.netsupervisor.service.InfoUserService"
    url="dubbo://192.168.139.129:20880"/>
复制代码

本文将重点分析通过注册中心方式，服务引用的过程。开始之前，我们再回顾一下消费者端项目整体的XML文件配置。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
       xmlns="http://www.springframework.org/schema/beans"
       xsi:schemaLocation="http://www.springframework.org/schema/beans 
	   http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
       http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
	<!-- 消费者方应用信息，用于计算依赖关系 -->
    <dubbo:application name="dubbo_consumer"/>
	<!-- 用于配置连接注册中心相关信息 -->
    <dubbo:registry protocol="zookeeper" address="192.168.139.131:2181" client="zkclient" />
	<!-- 引用配置 	用于创建一个远程接口服务代理 -->
    <dubbo:reference id="infoUserService"
    	interface="com.viewscenes.netsupervisor.service.InfoUserService"/>
</beans>
复制代码

一、引用的是什么？

在上述配置文件中， infoUserService 只是Spring中的一个Bean。在代码中，我们通过 getBean() 来获取它，然后调用它的方法。

我们在 RPC基本原理以及如何用Netty来实现RPC 这篇文章中已经分析过，这里的bean其实是一个 FactoryBean ，通过它可以返回一个接口的代理对象，完成调用逻辑的处理。

在代码中，我们通过这样来获取这个Bean。

InfoUserService infoUserService = (InfoUserService) context.getBean("infoUserService");
复制代码

这里的 infoUserService 就是一个代理对象，比如像 proxy@2903 这种。当然了，它必然会包含一个 InvocationHandler ，如下所示：

服务引用就是其实通过动态代理给接口创建代理对象并返回。当我们调用接口方法时，则调用到 InvocationHandler 相关方法，处理相关请求。

二、从哪里开始？

在上面的配置文件中， dubbo:reference 对应的是 ReferenceBean 处理类。那么Spring在实例化这个Bean的时候，就调用到里面方法。我们先看看它的类结构

package com.alibaba.dubbo.config.spring;

public class ReferenceBean<T> extends ReferenceConfig<T> implements 
		FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {

    private transient ApplicationContext applicationContext;

    public ReferenceBean() {
        super();
    }
    public ReferenceBean(Reference reference) {
        super(reference);
    }
    //设置applicationContext
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }
    //返回代理对象
    public Object getObject() throws Exception {
        return get();
    }
    //返回代理对象接口类型
    public Class<?> getObjectType() {
        return getInterfaceClass();
    }
	//是否为单例
    @Parameter(excluded = true)
    public boolean isSingleton() {
        return true;
    }
    //Bean初始化方法
    public void afterPropertiesSet() throws Exception {
        if(isInit()){
			getObject();
		}
    }
}
复制代码

首先，映入眼帘的是 FactoryBean 接口。有了它，则证明当前bean是一个工厂Bean，所以我们重点关注它的 getObject 方法，当我们的服务被注入到其他类中时，就会调用到此方法。

其次，是 InitializingBean 接口。它是Bean的初始化方法，当Bean完成实例化之后被调用。在上述代码中，当 init() 条件判断成立后调用 getObject 。我们可以在配置文件中以这种方式来激活它： init="true"

以上就是服务引用的两个不同时机。再专业点来讲，一个是懒汉式，一个是饿汉式。Dubbo默认是懒汉式引用，需要时才会调用。

上述代码中， getObject 方法会调用到父类的 init 方法。这个方法内容比较多，前面的部分是各种配置检查、赋值，然后就是创建代理对象返回。

private void init() {
	//避免重复初始化
	if (initialized) {
		return;
	}	
	//省略相关配置检查、赋值等代码...	
	
	//创建代理
	ref = createProxy(map);
}
复制代码

上述代码中，重点是 createProxy 方法。它负责创建Invoker实例和创建代理对象。

也许我们还记得，在服务暴露的时候，会分为本地暴露和远程暴露。在这里，服务引用也是这样。Dubbo首先判断服务引用是本地引用还是远程引用，默认是远程引用。然后判断是否为直连服务，根据协议调用 refer 方法创建invoker对象。最后创建服务代理对象并返回。

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) {
			isJvmRefer = false;
		} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
			// by default, reference local service if there is
			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 {
		//url不为空则代表是服务直连
		if (url != null && url.length() > 0) {
			//切割多个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.isEmpty()) {
				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.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; 
				}
			}
			if (registryURL != null) {
				// 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 { 
				invoker = cluster.join(new StaticDirectory(invokers));
			}
		}
	}
	//创建代理类
	return (T) proxyFactory.getProxy(invoker);
}
复制代码

前面的整个流程我们可以分为两大部分：创建Invoker和创建代理。

三、创建Invoker

1、连接注册中心

我们以本文开头的XML配置文件为例，它是基于单个注册中心的远程服务引用。所以，我们重点分析 invoker = refprotocol.refer(interfaceClass, urls.get(0));

我们需要清楚的认识到， urls.get(0) 获取到的是注册中心的地址。内容如下： registry://192.168.139.131:2181/org.apache.dubbo.registry.RegistryService?...

所以，我们先来看 RegistryProtocol.refer 方法。它主要是获取注册中心，并执行服务引用。

public class RegistryProtocol implements Protocol {
	
	public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
		//转换协议
		url = url.setProtocol(url.getParameter("registry", "dubbo")).removeParameter("registry");
		//通过zookeeper连接到注册中心
		Registry registry = registryFactory.getRegistry(url);
		if (RegistryService.class.equals(type)) {
			return proxyFactory.getInvoker((T) registry, type, url);
		}
		// group="a,b" or group="*"
		Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded("refer"));
		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);
			}
		}
		return doRefer(cluster, registry, type, url);
    }
}
复制代码

我们接着看 doRefer(cluster, registry, type, url); 我们可以认为它主要做了3件事：

向注册中心注册服务消费者(本身)
向注册中心订阅服务提供者
通过Cluster创建Invoker对象

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);
	Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
	//创建订阅 URL
	URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.
		remove("register.ip"), 0, type.getName(), parameters);
	if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
			&& url.getParameter(Constants.REGISTER_KEY, true)) {
		//向注册中心注册服务消费者
		registry.register(subscribeUrl.addParameters("category", "consumers",
				"check", 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.registerConsumer(invoker, url, subscribeUrl, directory);
	return invoker;
}
复制代码

如上代码，它的重点在于 directory.subscribe ，它通过订阅注册中心的服务提供者，调用相关协议的引用方法。比如 DubboProtocol.refer ，下面我们来看它是怎么调用的。

2、订阅服务提供者

我们通过 directory.subscribe 这句代码，会定位到 ZookeeperRegistry.doSubscribe 方法。

public class ZookeeperRegistry extends FailbackRegistry {
	
	protected void doSubscribe(final URL url, final NotifyListener listener) {
        
		List<URL> urls = new ArrayList<URL>();
		//获取注册中心除了消费者信息之外的所有节点
		//比如configurators, routers, providers
		String[] paths = toCategoriesPath(url);
		for (String path :paths) {
			//获取节点监听器管理容器
			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() {
					@Override
					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 数据变化，通知到zkListener
			//即通知到ZookeeperRegistry.this.notify
			List<String> children = zkClient.addChildListener(path, zkListener);
			if (children != null) {
				urls.addAll(toUrlsWithEmpty(url, path, children));
			}
		}
		//通知
		notify(url, listener, urls);  
    }
}
复制代码

如上代码，重点就是通过注册中心获取到服务提供者的url，调用notify方法。同时还订阅了服务提供者的节点数据变化，如果有变化也调用notify方法。而在notify方法中，最终会根据url调用服务引用方法refer，再封装成InvokerDelegate对象。 invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);

这样不仅完成了本次的服务引用，而且当服务提供者上下线的时候，消费者也可以动态感知。

3、Listener和Filter

以DubboProtocol为例，它是Wrapper包装类的对象，关于这一点我们一定要牢记。

回忆服务暴露的过程，我们可以监听服务暴露的过程，在构建调用链的时候可以增加Filter。在这里，服务引用也是一样的。

服务引用监听

在InvokerListener 接口中，有两个方法。分别是服务引用后调用和服务销毁时调用。

@SPI
public interface InvokerListener {	
	//服务引用
    void referred(Invoker<?> invoker) throws RpcException;
	//服务销毁
    void destroyed(Invoker<?> invoker);
}
复制代码

我们先自定义监听器，来实现接口的方法。

public class MyInvokerListener1 implements InvokerListener{

	public void referred(Invoker<?> invoker) throws RpcException {		
		System.out.println("服务引用111.."+invoker.getUrl());
	}
	public void destroyed(Invoker<?> invoker) {
		System.out.println("服务销毁111.."+invoker.getUrl());
	}
}
复制代码

然后创建扩展点配置文件，文件名称为： com.alibaba.dubbo.rpc.InvokerListener 内容为： listener1=com.viewscenes.netsupervisor.listener.MyInvokerListener1

最后，修改XML配置文件 <dubbo:consumer listener="listener1"/>

Filter过滤器

首先自定义过滤器

public class MyFilter implements Filter{

	public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
		System.out.println("调用之前-----------");
		Result result = invoker.invoke(invocation);
		System.out.println("调用之后-----------");
		return result;
	}
}
复制代码

然后创建扩展点配置文件，文件名称为： com.alibaba.dubbo.rpc.Filter 内容为： myfilter1=com.viewscenes.netsupervisor.filter.MyFilter

最后，修改XML配置文件，说明自定义的过滤器在默认过滤器之前。 filter="myfilter1,default""/>

4、DubboProtocol

前面啰啰嗦嗦一大堆，终于来到正主。这里主要是获取通信客户端，拿到ExchangeClient，然后封装为DubboInvoker对象并返回。

public class DubboProtocol extends AbstractProtocol {
	
	public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        optimizeSerialization(url);
		//获取通信客户端
		ExchangeClient[] client = getClients(url);
		//创建DubboInvoker对象
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, client, invokers);
        invokers.add(invoker);
        return invoker;
    }
}
复制代码

这里主要分为两个步骤：

创建通信客户端

在getClients方法里面，主要判断是否为共享连接的客户端。

private ExchangeClient[] getClients(URL url) {
	// 是否共享连接
	boolean service_share_connect = false;
	// 获取连接数，默认为0，表示未配置
	int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
	// 如果未配置 connections，则共享连接
	if (connections == 0) {
		service_share_connect = true;
		connections = 1;
	}
	ExchangeClient[] clients = new ExchangeClient[connections];
	for (int i = 0; i < clients.length; i++) {
		if (service_share_connect) {
			// 获取共享客户端
			clients[i] = getSharedClient(url);
		} else {
			//初始化新的客户端
			clients[i] = initClient(url);
		}
	}
	return clients;
}
复制代码

如果未配置，Dubbo默认创建共享连接客户端。我们继续看getSharedClient方法。

private ExchangeClient getSharedClient(URL url) {
	//服务地址
	String key = url.getAddress();
	//查询缓存是否已经创建了客户端
	ReferenceCountExchangeClient client = referenceClientMap.get(key);
	if (client != null) {
		if (!client.isClosed()) {
			client.incrementAndGetCount();
			return client;
		} else {
			referenceClientMap.remove(key);
		}
	}
	synchronized (key.intern()) {
		//初始化客户端
		ExchangeClient exchangeClient = initClient(url);
		//封装成ReferenceCountExchangeClient对象，用于引用计数
		client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
		//放入缓存并返回
		referenceClientMap.put(key, client);
		ghostClientMap.remove(key);
		return client;
	}
}
复制代码

如上代码，先从缓存中获取客户端，未命中则调用initClient方法初始化客户端。并将它封装成ReferenceCountExchangeClient 对象返回。在initClient方法中，设置客户端类型、编解码和心跳信息，并调用connect创建客户端。

private ExchangeClient initClient(URL url) {

	//客户端类型 默认为netty
	String str = url.getParameter("client", url.getParameter("server","netty"));
	//编解码
	url = url.addParameter("codes", "dubbo");
	//心跳
	url = url.addParameterIfAbsent("heartbeat","60000");

	ExchangeClient client;
	try {
		if (url.getParameter("lazy", false)) {
			client = new LazyConnectExchangeClient(url, requestHandler);
		} else {
			//创建连接客户端
			client = Exchangers.connect(url, requestHandler);
		}
	} catch (RemotingException e) {
		throw new RpcException("Fail to create remoting 
			client for service(" + url + "): " + e.getMessage(), e);
	}
	return client;
}
复制代码

如上代码，connect会调用到HeaderExchangeClient.connect方法。

public class HeaderExchanger implements Exchanger {

	public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
        HeaderExchangeHandler headerExchangeHandler = new HeaderExchangeHandler(handler);
        DecodeHandler decodeHandler = new DecodeHandler(headerExchangeHandler);
        Client client = Transporters.connect(url, decodeHandler);
        return new HeaderExchangeClient(client,true);
    }
}
复制代码

我们重点来看 Transporters.connect ，它会调用到 NettyClient 的构造方法。最终调用doOpen和doConnect方法，连接通信服务器。

public class NettyClient extends AbstractClient {

	protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();
        bootstrap = new ClientBootstrap(channelFactory);
        bootstrap.setOption("keepAlive", true);
        bootstrap.setOption("tcpNoDelay", true);
        bootstrap.setOption("connectTimeoutMillis", getTimeout());
		
		//处理器
        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() {
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), 
						getUrl(), NettyClient.this);
                ChannelPipeline pipeline = Channels.pipeline();
                pipeline.addLast("decoder", adapter.getDecoder());
                pipeline.addLast("encoder", adapter.getEncoder());
                pipeline.addLast("handler", nettyHandler);
                return pipeline;
            }
        });
    }
	
	
	protected void doConnect() throws Throwable {
        long start = System.currentTimeMillis();
		//获取服务器地址，并连接
        ChannelFuture future = bootstrap.connect(getConnectAddress());
        try {
            boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
			//连接成功后获取Channel
            if (ret && future.isSuccess()) {
                Channel newChannel = future.getChannel();
                newChannel.setInterestOps(Channel.OP_READ_WRITE);
                try {
                    // Close old channel
                    Channel oldChannel = NettyClient.this.channel; // copy reference
                    if (oldChannel != null) {
                        try {                  
                            oldChannel.close();
                        } finally {
                            NettyChannel.removeChannelIfDisconnected(oldChannel);
                        }
                    }
                } finally {
                    if (NettyClient.this.isClosed()) {
                        try {                           
                            newChannel.close();
                        } finally {
                            NettyClient.this.channel = null;
                            NettyChannel.removeChannelIfDisconnected(newChannel);
                        }
                    } else {
						//设置当前Channel
                        NettyClient.this.channel = newChannel;
                    }
                }
            }
        }
    }
}
复制代码

以上大部分都是Netty的代码，连接服务器成功后获取到Channel保存。我们重点看 NettyHandler ，它是事件处理器。这里的 messageReceived 负责接收处理返回的消息。接收到消息后，通过线程池拿到线程去执行后续的处理。不同的是，这里的线程池类型是CachedThreadPool ，即缓存型线程池。

这里的流程处理完毕，我们再把目光回到 DubboInvoker ，它通过一个构造方法返回Invoker对象。

public DubboInvoker(Class<T> serviceType, URL url, 
			ExchangeClient[] clients, Set<Invoker<?>> invokers) {
	this.clients = clients;
	this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
	this.invokers = invokers;
}
复制代码

至此，Invoker对象已经创建完毕。我们回到开头的 createProxy 方法。最后一步是为这个Invoker对象创建代理。

五、创建代理

在Dubbo，默认使用Javassist来创建代理。

public class JavassistProxyFactory extends AbstractProxyFactory {

    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
		//调用处理程序
		InvocationHandler invocationHandler = new InvokerInvocationHandler(invoker)
		//创建代理
        return (T) Proxy.getProxy(interfaces).newInstance(invocationHandler);
    }
}
复制代码

创建的过程很简单，我们主要来看 InvocationHandler 。既然是代理对象，我们每次调用接口方法的时候，会调用到这个对象的 invoke 方法。

public class InvokerInvocationHandler implements InvocationHandler {

	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]);
        }
		//把请求的方法名和参数等封装成RpcInvocation对象
		//调用invoke
        return invoker.invoke(new RpcInvocation(method, args)).recreate();
    }
}
复制代码

至此，服务引用的大体流程我们已经分析完了。我们再来总结一下它的过程：