dubbo服务引用初始化过程就是创建服务动态代理的过程，与服务发布一样，同样借助bean初始化完成动态代理的创建。具体调用过程：
com.alibaba.dubbo.config.ReferenceConfig：
get()-->init()-->createProxy()
在createProxy()方法中，服务引用分为三种情况：
1.JVM内部引用的代理
2.用户指定URL 直连
3.通过注册中心，引用远程的代理
直接看第三种
private T createProxy(Map<String, String> map) {
.....................................
List<URL> us = loadRegistries(false);//导入注册中心url
if (us != null && us.size() > 0) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);//导入monitor相关信息，加入URL中
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
// 多个注册中心，封装成一个Invoker
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);
}
主要服务引用过程在这个方法里refprotocol.refer(interfaceClass, url)，调用到RegistryProtocol.refer()，根据url获取到注册中心，然后调用doRefer(),new 一个RegistryDirectory，主要缓存远程服务相关信息，如ip:port，然后向注册中心consumer端注册自己，然后在相应节点providers端订阅提供者的信息，具体过程如下图
clipboard.png
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
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)) {//注册自己
registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
Constants.CHECK_KEY, String.valueOf(false)));
}
//订阅provider注册中心节点信息，此处url需要携带providers参数
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
return cluster.join(directory);
}
下面进入directory.subscribe()方法，此方法非常重要，此处完成提供者信息的获取，通信客户端的初始化，此方法调用链为：FailbackRegistry.subscribe()-->ZookeeperRegistry.doSubscribe()-->RegistryDirectory.notify(),下面来看ZookeeperRegistry.doSubscribe()方法，此处主要完成提供端信息的获取，转换成url，订阅的节点有三类：
/dubbo/xx.xx/providers;
/dubbo/xx.xx/routes;
/dubbo/xx.xx/configurations;
protected void doSubscribe(final URL url, final NotifyListener listener) {
List<URL> urls = new ArrayList<URL>();//缓存订阅到信息
for (String path : toCategoriesPath(url)) {
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
...............................................
zkClient.create(path, false);
//从zk订阅到的信息,并添加子节点监听
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
notify(url, listener, urls);
}
然后调用到AbstractRegistry.notify()方法
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
..........................................
for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
String category = entry.getKey();
List<URL> categoryList = entry.getValue();
categoryNotified.put(category, categoryList);
saveProperties(url);//url信息保存到本地
listener.notify(categoryList);
}
}
listener就是一路传过来的RegistryDirectory，进入RegistryDirectory.notify()-->refreshInvoker()
private void refreshInvoker(List<URL> invokerUrls){
..........................................
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
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);
}
}
上面的方法提供者url信息转换成invoker，并与method对应起来。下面进入toInvokers(invokerUrls)：
private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
//检查提供者相关信息.......
....................
// 缓存key为没有合并消费端参数的URL，不管消费端如何合并参数，如果服务端URL发生变化，则重新refer
Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
if (invoker == null) { // 缓存中没有，重新refer
try {
........................................
if (enabled) {
invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
}
} catch (Throwable t) {
logger.error("Failed to refer invoker for interface:"+serviceType+",url:("+url+")" + t.getMessage(), t);
}
if (invoker != null) { // 将新的引用放入缓存
newUrlInvokerMap.put(key, invoker);
}
}else {
newUrlInvokerMap.put(key, invoker);
}
}
keys.clear();
return newUrlInvokerMap;
}
下面调用protocol.refer(serviceType, url)，此protocol为适配类，调用过程为ProtocolFilterWrapper.refer()-->ProtocolListenerWrapper.refer()-->DubboProtocol.refer():
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
// create rpc invoker.
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
然后调用getClients(url)，创建ExchangeClient用来处理和提供端的链接，一般一个服务接口只会创建一个client
private ExchangeClient[] getClients(URL url){
//是否共享连接
boolean service_share_connect = false;
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;
}
然后一路调用
Exchangers.connect()-->HeaderExchanger.connect()->Transporters.connect()-->NettyTransporter.connect()，最终初始化一个NettyClient，调用doopen()方法，完成网络客户端的初始化
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
bootstrap = new ClientBootstrap(channelFactory);
// config
// @see org.jboss.netty.channel.socket.SocketChannelConfig
bootstrap.setOption("keepAlive", true);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("connectTimeoutMillis", getTimeout());
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
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;
}
});
}
回到RegistryProtocol.doRefer()中，cluster.join(directory)，此处调用过程为：MockClusterWrapper.join()-->FailfastCluster.join()，最终初始化一个MockClusterInvoker对象，此对象持有RegistryDirectory和FailfastClusterInvoker的引用，接下来调用JavassistProxyFactory.getProxy()创建代理：
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
此对象spring注入服务接口中，形成一个透明化的本地服务，当调用本地接口方法时，就会启动动态代理，执行InvokerInvocationHandler.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]);
}
//调用MockClusterInvoker.invoke()方法
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
