在前面的章节中,我们已经完成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和创建代理。
我们以本文开头的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件事:
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
,下面我们来看它是怎么调用的。
我们通过 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);
这样不仅完成了本次的服务引用,而且当服务提供者上下线的时候,消费者也可以动态感知。
以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"/>
首先自定义过滤器
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""/>
前面啰啰嗦嗦一大堆,终于来到正主。这里主要是获取通信客户端,拿到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(); } } 复制代码
至此,服务引用的大体流程我们已经分析完了。我们再来总结一下它的过程: