杭州定制网站开发,荣成建设局网站,网站店招用什么软件做的,建设什么网站1. 简介 在上一篇文章中#xff0c;我详细的分析了服务导出的原理。本篇文章我们趁热打铁#xff0c;继续分析服务引用的原理。在 Dubbo 中#xff0c;我们可以通过两种方式引用远程服务。第一种是使用服务直联的方式引用服务#xff0c;第二种方式是基于注册中心进行引用。… 1. 简介 在上一篇文章中我详细的分析了服务导出的原理。本篇文章我们趁热打铁继续分析服务引用的原理。在 Dubbo 中我们可以通过两种方式引用远程服务。第一种是使用服务直联的方式引用服务第二种方式是基于注册中心进行引用。服务直联的方式仅适合在调试或测试服务的场景下使用不适合在线上环境使用。因此本文我将重点分析通过注册中心引用服务的过程。从注册中心中获取服务配置只是服务引用过程中的一环除此之外服务消费者还需要经历 Invoker 创建、代理类创建等步骤。这些步骤我将在后续章节中一一进行分析。 2.服务引用原理 Dubbo 服务引用的时机有两个第一个是在 Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时引用服务第二个是在 ReferenceBean 对应的服务被注入到其他类中时引用。这两个引用服务的时机区别在于第一个是饿汉式的第二个是懒汉式的。默认情况下Dubbo 使用懒汉式引用服务。如果需要使用饿汉式可通过配置 dubbo:reference 的 init 属性开启。下面我们按照 Dubbo 默认配置进行分析整个分析过程从 ReferenceBean 的 getObject 方法开始。当我们的服务被注入到其他类中时Spring 会第一时间调用 getObject 方法并由该方法执行服务引用逻辑。按照惯例在进行具体工作之前需先进行配置检查与收集工作。接着根据收集到的信息决定服务用的方式有三种第一种是引用本地 (JVM) 服务第二是通过直联方式引用远程服务第三是通过注册中心引用远程服务。不管是哪种引用方式最后都会得到一个 Invoker 实例。如果有多个注册中心多个服务提供者这个时候会得到一组 Invoker 实例此时需要通过集群管理类 Cluster 将多个 Invoker 合并成一个实例。合并后的 Invoker 实例已经具备调用本地或远程服务的能力了但并不能将此实例暴露给用户使用这会对用户业务代码造成侵入。此时框架还需要通过代理工厂类 (ProxyFactory) 为服务接口生成代理类并让代理类去调用 Invoker 逻辑。避免了 Dubbo 框架代码对业务代码的侵入同时也让框架更容易使用。 以上就是 Dubbo 引用服务的大致原理下面我们深入到代码中详细分析服务引用细节。 3.源码分析 服务引用的入口方法为 ReferenceBean 的 getObject 方法该方法定义在 Spring 的 FactoryBean 接口中ReferenceBean 实现了这个方法。实现代码如下 public Object getObject() throws Exception {return get();
}public synchronized T get() {if (destroyed) {throw new IllegalStateException(Already destroyed!);}// 检测 ref 是否为空为空则通过 init 方法创建if (ref null) {// init 方法主要用于处理配置以及调用 createProxy 生成代理类init();}return ref;
} 这里两个方法代码都比较简短并不难理解。不过这里需要特别说明一下如果大家从 getObject 方法进行代码调试时会碰到比较诧异的问题。这里假设你使用 IDEA且保持了 IDEA 的默认配置。当你面调试到 get 方法的if (ref null)时你会惊奇的发现 ref 不为空导致你无法进入到 init 方法中继续调试。导致这个现象的原因是 Dubbo 框架本身有点小问题这个小问题会引发一些让人诧异的现象。关于这个问题我进行了将近两个小时的排查。查明问题后我给 Dubbo 提交了一个 pull request (#2754 ) 介绍这个问题有兴趣的朋友可以去看看。大家如果想规避这个问题可以修改一下 IDEA 的配置。在配置面板中搜索 toString然后取消Enable toString object view前的对号。具体如下 讲完需要注意的点我们继续向下分析接下来将分析配置的处理过程。 3.1 处理配置 Dubbo 提供了丰富的配置用于调整和优化框架行为性能等。Dubbo 在引用或导出服务时首先会对这些配置进行检查和处理以保证配置到正确性。如果大家不是很熟悉 Dubbo 配置建议先阅读以下官方文档。配置解析的方法为 ReferenceConfig 的 init 方法下面来看一下方法逻辑。 private void init() {if (initialized) {return;}initialized true;if (interfaceName null || interfaceName.length() 0) {throw new IllegalStateException(interface not allow null!);}// 检测 consumer 变量是否为空为空则创建checkDefault();appendProperties(this);if (getGeneric() null getConsumer() ! null) {// 设置 genericsetGeneric(getConsumer().getGeneric());}// 检测是否为泛化接口if (ProtocolUtils.isGeneric(getGeneric())) {interfaceClass GenericService.class;} else {try {// 加载类interfaceClass Class.forName(interfaceName, true, Thread.currentThread().getContextClassLoader());} catch (ClassNotFoundException e) {throw new IllegalStateException(e.getMessage(), e);}checkInterfaceAndMethods(interfaceClass, methods);}// ------------------------------- 分割线1 ------------------------------// 从系统变量中获取与接口名对应的属性值String resolve System.getProperty(interfaceName);String resolveFile null;if (resolve null || resolve.length() 0) {// 从系统属性中获取解析文件路径resolveFile System.getProperty(dubbo.resolve.file);if (resolveFile null || resolveFile.length() 0) {// 从指定位置加载配置文件File userResolveFile new File(new File(System.getProperty(user.home)), dubbo-resolve.properties);if (userResolveFile.exists()) {// 获取文件绝对路径resolveFile userResolveFile.getAbsolutePath();}}if (resolveFile ! null resolveFile.length() 0) {Properties properties new Properties();FileInputStream fis null;try {fis new FileInputStream(new File(resolveFile));// 从文件中加载配置properties.load(fis);} catch (IOException e) {throw new IllegalStateException(Unload ..., cause:...);} finally {try {if (null ! fis) fis.close();} catch (IOException e) {logger.warn(e.getMessage(), e);}}// 获取与接口名对应的配置resolve properties.getProperty(interfaceName);}}if (resolve ! null resolve.length() 0) {// 将 resolve 赋值给 urlurl resolve;}// ------------------------------- 分割线2 ------------------------------if (consumer ! null) {if (application null) {// 从 consumer 中获取 Application 实例下同application consumer.getApplication();}if (module null) {module consumer.getModule();}if (registries null) {registries consumer.getRegistries();}if (monitor null) {monitor consumer.getMonitor();}}if (module ! null) {if (registries null) {registries module.getRegistries();}if (monitor null) {monitor module.getMonitor();}}if (application ! null) {if (registries null) {registries application.getRegistries();}if (monitor null) {monitor application.getMonitor();}}// 检测本地 Application 和本地存根配置合法性checkApplication();checkStubAndMock(interfaceClass);// ------------------------------- 分割线3 ------------------------------MapString, String map new HashMapString, String();MapObject, Object attributes new HashMapObject, Object();// 添加 side、协议版本信息、时间戳和进程号等信息到 map 中map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);map.put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));if (ConfigUtils.getPid() 0) {map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));}if (!isGeneric()) { // 非泛化服务// 获取版本String revision Version.getVersion(interfaceClass, version);if (revision ! null revision.length() 0) {map.put(revision, revision);}// 获取接口方法列表并添加到 map 中String[] methods Wrapper.getWrapper(interfaceClass).getMethodNames();if (methods.length 0) {map.put(methods, Constants.ANY_VALUE);} else {map.put(methods, StringUtils.join(new HashSetString(Arrays.asList(methods)), ,));}}map.put(Constants.INTERFACE_KEY, interfaceName);// 将 ApplicationConfig、ConsumerConfig、ReferenceConfig 等对象的字段信息添加到 map 中appendParameters(map, application);appendParameters(map, module);appendParameters(map, consumer, Constants.DEFAULT_KEY);appendParameters(map, this);// ------------------------------- 分割线4 ------------------------------String prefix StringUtils.getServiceKey(map);if (methods ! null !methods.isEmpty()) {// 遍历 MethodConfig 列表for (MethodConfig method : methods) {appendParameters(map, method, method.getName());String retryKey method.getName() .retry;// 检测 map 是否包含 methodName.retryif (map.containsKey(retryKey)) {String retryValue map.remove(retryKey);if (false.equals(retryValue)) {// 添加重试次数配置 methodName.retriesmap.put(method.getName() .retries, 0);}}// 添加 MethodConfig 中的“属性”字段到 attributes// 比如 onreturn、onthrow、oninvoke 等appendAttributes(attributes, method, prefix . method.getName());checkAndConvertImplicitConfig(method, map, attributes);}}// ------------------------------- 分割线5 ------------------------------// 获取服务消费者 ip 地址String hostToRegistry ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);if (hostToRegistry null || hostToRegistry.length() 0) {hostToRegistry NetUtils.getLocalHost();} else if (isInvalidLocalHost(hostToRegistry)) {throw new IllegalArgumentException(Specified invalid registry ip from property... );}map.put(Constants.REGISTER_IP_KEY, hostToRegistry);// 存储 attributes 到系统上下文中StaticContext.getSystemContext().putAll(attributes);// 创建代理类ref createProxy(map);// 根据服务名ReferenceConfig代理类构建 ConsumerModel// 并将 ConsumerModel 存入到 ApplicationModel 中ConsumerModel consumerModel new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
} 上面的代码很长做的事情比较多。这里我根据代码逻辑对代码进行了分块下面我们一起来看一下。 首先是方法开始到分割线1之间的代码。这段代码主要用于检测 ConsumerConfig 实例是否存在如不存在则创建一个新的实例然后通过系统变量或 dubbo.properties 配置文件填充 ConsumerConfig 的字段。接着是检测泛化配置并根据配置设置 interfaceClass 的值。本段代码逻辑大致就是这些接着来看分割线1到分割线2之间的逻辑。这段逻辑用于从系统属性或配置文件中加载与接口名相对应的配置并将解析结果赋值给 url 字段。url 字段的作用一般是用于点对点调用。继续向下看分割线2和分割线3之间的代码用于检测几个核心配置类是否为空为空则尝试从其他配置类中获取。分割线3与分割线4之间的代码主要是用于收集各种配置并将配置存储到 map 中。分割线4和分割线5之间的代码用于处理 MethodConfig 实例。该实例包含了事件通知配置比如 onreturn、onthrow、oninvoke 等。分割线5到方法结尾的代码主要用于解析服务消费者 ip以及调用 createProxy 创建代理对象。关于该方法的详细分析将会在接下来的章节中展开。 到这里关于配置的检查与处理过长就分析完了。这部分逻辑不是很难理解但比较繁杂大家需要耐心看一下。好了本节先到这接下来分析服务引用的过程。 3.2 引用服务 本节我们要从 createProxy 开始看起。createProxy 这个方法表面上看起来只是用于创建代理对象但实际上并非如此。该方法还会调用其他方法构建以及合并 Invoker 实例。具体细节如下。 private T createProxy(MapString, String map) {URL tmpUrl new URL(temp, localhost, 0, map);final boolean isJvmRefer;if (isInjvm() null) {// url 配置被指定则不做本地引用if (url ! null url.length() 0) {isJvmRefer false;// 根据 url 的协议、scope 以及 injvm 等参数检测是否需要本地引用// 比如如果用户显式配置了 scopelocal此时 isInjvmRefer 返回 true} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {isJvmRefer true;} else {isJvmRefer false;}} else {// 获取 injvm 配置值isJvmRefer isInjvm().booleanValue();}// 本地引用if (isJvmRefer) {// 生成本地引用 URL协议为 injvmURL url new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);// 调用 refer 方法构建 InjvmInvoker 实例invoker refprotocol.refer(interfaceClass, url);// 远程引用} 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 url.setPath(interfaceName);}// 检测 url 协议是否为 registry若是表明用户想使用指定的注册中心if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {// 将 map 转换为查询字符串并作为 refer 参数的值添加到 url 中urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));} else {// 合并 url移除服务提供者的一些配置这些配置来源于用户配置的 url 属性// 比如线程池相关配置。并保留服务提供者的部分配置比如版本group时间戳等// 最后将合并后的配置设置为 url 查询字符串中。urls.add(ClusterUtils.mergeUrl(url, map));}}}} else {// 加载注册中心 urlListURL 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()));}// 添加 refer 参数到 url 中并将 url 添加到 urls 中urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));}}// 未配置注册中心抛出异常if (urls.isEmpty()) {throw new IllegalStateException(No such any registry to reference...);}}// 单个注册中心或服务提供者(服务直联下同)if (urls.size() 1) {// 调用 RegistryProtocol 的 refer 构建 Invoker 实例invoker refprotocol.refer(interfaceClass, urls.get(0));// 多个注册中心或多个服务提供者或者两者混合} else {ListInvoker? invokers new ArrayListInvoker?();URL registryURL null;// 获取所有的 Invokerfor (URL url : urls) {// 通过 refprotocol 调用 refer 构建 Invokerrefprotocol 会在运行时// 根据 url 协议头加载指定的 Protocol 实例并调用实例的 refer 方法invokers.add(refprotocol.refer(interfaceClass, url));if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {registryURL url;}}if (registryURL ! null) {// 如果注册中心链接不为空则将使用 AvailableClusterURL u registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);// 创建 StaticDirectory 实例并由 Cluster 对多个 Invoker 进行合并invoker cluster.join(new StaticDirectory(u, invokers));} else {invoker cluster.join(new StaticDirectory(invokers));}}}Boolean c check;if (c null consumer ! null) {c consumer.isCheck();}if (c null) {c true;}// invoker 可用性检查if (c !invoker.isAvailable()) {throw new IllegalStateException(No provider available for the service...);}// 生成代理类return (T) proxyFactory.getProxy(invoker);
} 上面代码很多不过逻辑比较清晰。首先根据配置检查是否为本地调用若是则调用 InjvmProtocol 的 refer 方法生成 InjvmInvoker 实例。若不是则读取直联配置项或注册中心 url并将读取到的 url 存储到 urls 中。然后根据 urls 元素数量进行后续操作。若 urls 元素数量为1则直接通过 Protocol 自适应拓展构建 Invoker 实例接口。若 urls 元素数量大于1即存在多个注册中心或服务直联 url此时先根据 url 构建 Invoker。然后再通过 Cluster 合并多个 Invoker最后调用 ProxyFactory 生成代理类。这里Invoker 的构建过程以及代理类的过程比较重要因此我将分两小节对这两个过程进行分析。 3.2.1 创建 Invoker Invoker 是 Dubbo 的核心模型代表一个可执行体。在服务提供方Invoker 用于调用服务提供类。在服务消费方Invoker 用于执行远程调用。Invoker 在 Dubbo 中的位置十分重要因此我们有必要去搞懂它。从前面的代码中可知Invoker 是由 Protocol 实现类构建的。Protocol 实现类有很多这里我会分析最常用的两个分别是 RegistryProtocol 和 DubboProtocol其他的大家自行分析。下面先来分析 DubboProtocol 的 refer 方法源码。如下 public T InvokerT refer(ClassT serviceType, URL url) throws RpcException {optimizeSerialization(url);// 创建 DubboInvokerDubboInvokerT invoker new DubboInvokerT(serviceType, url, getClients(url), invokers);invokers.add(invoker);return invoker;
} 上面方法看起来比较简单不过这里有一个调用需要我们注意一下即 getClients。这个方法用于获取客户端实例实例类型为 ExchangeClient。ExchangeClient 实际上并不具备通信能力因此它需要更底层的客户端实例进行通信。比如 NettyClient、MinaClient 等默认情况下Dubbo 使用 NettyClient 进行通信。接下来我们简单看一下 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;
} 这里根据 connections 数量决定是获取共享客户端还是创建新的客户端实例默认情况下使用共享客户端实例。不过 getSharedClient 方法中也会调用 initClient 方法因此下面我们一起看一下这两个方法。 private ExchangeClient getSharedClient(URL url) {String key url.getAddress();// 获取带有“引用计数”功能的 ExchangeClientReferenceCountExchangeClient client referenceClientMap.get(key);if (client ! null) {if (!client.isClosed()) {// 增加引用计数client.incrementAndGetCount();return client;} else {referenceClientMap.remove(key);}}locks.putIfAbsent(key, new Object());synchronized (locks.get(key)) {if (referenceClientMap.containsKey(key)) {return referenceClientMap.get(key);}// 创建 ExchangeClient 客户端ExchangeClient exchangeClient initClient(url);// 将 ExchangeClient 实例传给 ReferenceCountExchangeClient这里明显用了装饰模式client new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);referenceClientMap.put(key, client);ghostClientMap.remove(key);locks.remove(key);return client;}
} 上面方法先访问缓存若缓存未命中则通过 initClient 方法创建新的 ExchangeClient 实例并将该实例传给 ReferenceCountExchangeClient 构造方法创建一个带有引用技术功能的 ExchangeClient 实例。ReferenceCountExchangeClient 内部实现比较简单就不分析了。下面我们再来看一下 initClient 方法的代码。 private ExchangeClient initClient(URL url) {// 获取客户端类型默认为 nettyString str url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));// 添加编解码和心跳包参数到 url 中url url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);url url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));// 检测客户端类型是否存在不存在则抛出异常if (str ! null str.length() 0 !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {throw new RpcException(Unsupported client type: ...);}ExchangeClient client;try {// 获取 lazy 配置并根据配置值决定创建的客户端类型if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {// 创建懒加载 ExchangeClient 实例client new LazyConnectExchangeClient(url, requestHandler);} else {// 创建普通 ExchangeClient 实例client Exchangers.connect(url, requestHandler);}} catch (RemotingException e) {throw new RpcException(Fail to create remoting client for service...);}return client;
} initClient 方法首先获取用户配置的客户端类型默认为 netty。然后检测用户配置的客户端类型是否存在不存在则抛出异常。最后根据 lazy 配置决定创建什么类型的客户端。这里的 LazyConnectExchangeClient 代码并不是很复杂该类会在 request 方法被调用时通过 Exchangers 的 connect 方法创建 ExchangeClient 客户端这里就不分析 LazyConnectExchangeClient 的代码了。下面我们分析一下 Exchangers 的 connect 方法。 public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {if (url null) {throw new IllegalArgumentException(url null);}if (handler null) {throw new IllegalArgumentException(handler null);}url url.addParameterIfAbsent(Constants.CODEC_KEY, exchange);// 获取 Exchanger 实例默认为 HeaderExchangeClientreturn getExchanger(url).connect(url, handler);
} 如上getExchanger 会通过 SPI 加载 HeaderExchangeClient 实例这个方法比较简单大家自己看一下吧。接下来分析 HeaderExchangeClient 的实现。 public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {// 这里包含了多个调用分别如下// 1. 创建 HeaderExchangeHandler 对象// 2. 创建 DecodeHandler 对象// 3. 通过 Transporters 构建 Client 实例// 4. 创建 HeaderExchangeClient 对象return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
} 这里的调用比较多我们这里重点看一下 Transporters 的 connect 方法。如下 public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {if (url null) {throw new IllegalArgumentException(url null);}ChannelHandler handler;if (handlers null || handlers.length 0) {handler new ChannelHandlerAdapter();} else if (handlers.length 1) {handler handlers[0];} else {// 如果 handler 数量大于1则创建一个 ChannelHandler 分发器handler new ChannelHandlerDispatcher(handlers);}// 获取 Transporter 自适应拓展类并调用 connect 方法生成 Client 实例return getTransporter().connect(url, handler);
} 这里getTransporter 方法返回的是自适应拓展类该类会在运行时根据客户端类型加载指定的 Transporter 实现类。若用户未显示配置客户端类型则默认加载 NettyTransporter并调用该类的 connect 方法。如下 public Client connect(URL url, ChannelHandler listener) throws RemotingException {// 创建 NettyClient 对象return new NettyClient(url, listener);
} 到这里就不继续跟下去了在往下就是通过 Netty 提供的接口构建 Netty 客户端了大家有兴趣自己看看。到这里关于 DubboProtocol 的 refer 方法就分析完了。接下来继续分析 RegistryProtocol 的 refer 方法逻辑。 public T InvokerT refer(ClassT type, URL url) throws RpcException {// 取 registry 参数值并将其设置为协议头url url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);// 获取注册中心实例Registry registry registryFactory.getRegistry(url);// 这个判断暂时不知道有什么意图为什么要给 RegistryService 类型生成 Invoker if (RegistryService.class.equals(type)) {return proxyFactory.getInvoker((T) registry, type, url);}// 将 url 查询字符串转为 MapMapString, String qs StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));// 获取 group 配置String group qs.get(Constants.GROUP_KEY);if (group ! null group.length() 0) {if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length 1|| *.equals(group)) {// 通过 SPI 加载 MergeableCluster 实例并调用 doRefer 继续执行引用服务逻辑return doRefer(getMergeableCluster(), registry, type, url);}}// 调用 doRefer 继续执行引用服务逻辑return doRefer(cluster, registry, type, url);
} 上面代码首先为 url 设置协议头然后根据 url 参数加载注册中心实例。接下来对 RegistryService 继续针对性处理这个处理逻辑我不是很明白不知道为什么要为 RegistryService 类型生成 Invoker有知道同学麻烦告知一下。然后就是获取 group 配置根据 group 配置决定 doRefer 第一个参数的类型。这里的重点是 doRefer 方法如下 private T InvokerT doRefer(Cluster cluster, Registry registry, ClassT type, URL url) {// 创建 RegistryDirectory 实例RegistryDirectoryT directory new RegistryDirectoryT(type, url);// 设置注册中心和协议directory.setRegistry(registry);directory.setProtocol(protocol);MapString, String parameters new HashMapString, String(directory.getUrl().getParameters());// 生成服务消费者链接URL subscribeUrl new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);// 注册服务消费者在 consumers 目录下新节点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)));}// 订阅 providers、configurators、routers 等节点数据directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,Constants.PROVIDERS_CATEGORY , Constants.CONFIGURATORS_CATEGORY , Constants.ROUTERS_CATEGORY));// 一个注册中心可能有多个服务提供者因此这里需要将多个服务提供者合并为一个Invoker invoker cluster.join(directory);ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);return invoker;
} 如上doRefer 方法创建一个 RegistryDirectory 实例然后生成服务者消费者链接并向注册中心进行注册。注册完毕后紧接着订阅 providers、configurators、routers 等节点下的数据。完成订阅后RegistryDirectory 会收到这几个节点下的子节点信息比如可以获取到服务提供者的配置信息。由于一个服务可能部署在多台服务器上这样就会在 providers 产生多个节点这个时候就需要 Cluster 将多个服务节点合并为一个并生成一个 Invoker。关于 RegistryDirectory 和 Cluster本文不打算进行分析相关分析将会在随后的文章中展开。 好了关于 Invoker 的创建的逻辑就先分析到这。逻辑比较多大家耐心看一下。 3.2.2 创建代理 Invoker 创建完毕后接下来要做的事情是为服务接口生成代理对象。有了代理对象我们就可以通过代理对象进行远程调用。代理对象生成的入口方法为在 ProxyFactory 的 getProxy接下来进行分析。 public T T getProxy(InvokerT invoker) throws RpcException {// 调用重载方法return getProxy(invoker, false);
}public T T getProxy(InvokerT invoker, boolean generic) throws RpcException {Class?[] interfaces null;// 获取接口列表String config invoker.getUrl().getParameter(interfaces);if (config ! null config.length() 0) {// 切分接口列表String[] types Constants.COMMA_SPLIT_PATTERN.split(config);if (types ! null types.length 0) {interfaces new Class?[types.length 2];// 设置服务接口类和 EchoService.class 到 interfaces 中interfaces[0] invoker.getInterface();interfaces[1] EchoService.class;for (int i 0; i types.length; i) {// 加载接口类interfaces[i 1] ReflectUtils.forName(types[i]);}}}if (interfaces null) {interfaces new Class?[]{invoker.getInterface(), EchoService.class};}// 为 http 和 hessian 协议提供泛化调用支持参考 pull request #1827if (!invoker.getInterface().equals(GenericService.class) generic) {int len interfaces.length;Class?[] temp interfaces;// 创建新的 interfaces 数组interfaces new Class?[len 1];System.arraycopy(temp, 0, interfaces, 0, len);// 设置 GenericService.class 到数组中interfaces[len] GenericService.class;}// 调用重载方法return getProxy(invoker, interfaces);
}public abstract T T getProxy(InvokerT invoker, Class?[] types); 如上上面大段代码都是用来获取 interfaces 数组的因此我们需要继续往下看。getProxy(Invoker, Class?[]) 这个方法是一个抽象方法下面我们到 JavassistProxyFactory 类中看一下该方法的实现代码。 public T T getProxy(InvokerT invoker, Class?[] interfaces) {// 生成 Proxy 子类Proxy 是抽象类。并调用Proxy 子类的 newInstance 方法生成 Proxy 实例return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
} 上面代码并不多首先是通过 Proxy 的 getProxy 方法获取 Proxy 子类然后创建 InvokerInvocationHandler 对象并将该对象传给 newInstance 生成 Proxy 实例。InvokerInvocationHandler 实现自 JDK 的 InvocationHandler 接口具体的用途是拦截接口类调用。该类逻辑比较简单这里就不分析了。下面我们重点关注一下 Proxy 的 getProxy 方法如下。 public static Proxy getProxy(Class?... ics) {// 调用重载方法return getProxy(ClassHelper.getClassLoader(Proxy.class), ics);
}public static Proxy getProxy(ClassLoader cl, Class?... ics) {if (ics.length 65535)throw new IllegalArgumentException(interface limit exceeded);StringBuilder sb new StringBuilder();// 遍历接口列表for (int i 0; i ics.length; i) {String itf ics[i].getName();// 检测类型是否为接口if (!ics[i].isInterface())throw new RuntimeException(itf is not a interface.);Class? tmp null;try {// 重新加载接口类tmp Class.forName(itf, false, cl);} catch (ClassNotFoundException e) {}// 检测接口是否相同这里 tmp 有可能为空if (tmp ! ics[i])throw new IllegalArgumentException(ics[i] is not visible from class loader);// 拼接接口全限定名分隔符为 ;sb.append(itf).append(;);}// 使用拼接后接口名作为 keyString key sb.toString();MapString, Object cache;synchronized (ProxyCacheMap) {cache ProxyCacheMap.get(cl);if (cache null) {cache new HashMapString, Object();ProxyCacheMap.put(cl, cache);}}Proxy proxy null;synchronized (cache) {do {// 从缓存中获取 ReferenceProxy 实例Object value cache.get(key);if (value instanceof Reference?) {proxy (Proxy) ((Reference?) value).get();if (proxy ! null) {return proxy;}}// 多线程控制保证只有一个线程可以进行后续操作if (value PendingGenerationMarker) {try {// 其他线程在此处进行等待cache.wait();} catch (InterruptedException e) {}} else {// 放置标志位到缓存中并跳出 while 循环进行后续操作cache.put(key, PendingGenerationMarker);break;}}while (true);}long id PROXY_CLASS_COUNTER.getAndIncrement();String pkg null;ClassGenerator ccp null, ccm null;try {// 创建 ClassGenerator 对象ccp ClassGenerator.newInstance(cl);SetString worked new HashSetString();ListMethod methods new ArrayListMethod();for (int i 0; i ics.length; i) {// 检测接口访问级别是否为 protected 或 priveteif (!Modifier.isPublic(ics[i].getModifiers())) {// 获取接口包名String npkg ics[i].getPackage().getName();if (pkg null) {pkg npkg;} else {if (!pkg.equals(npkg))// 非 public 级别的接口必须在同一个包下否者抛出异常throw new IllegalArgumentException(non-public interfaces from different packages);}}// 添加接口到 ClassGenerator 中ccp.addInterface(ics[i]);// 遍历接口方法for (Method method : ics[i].getMethods()) {// 获取方法描述可理解为方法签名String desc ReflectUtils.getDesc(method);// 如果已包含在 worked 中则忽略。考虑这种情况// A 接口和 B 接口中包含一个完全相同的方法if (worked.contains(desc))continue;worked.add(desc);int ix methods.size();// 获取方法返回值类型Class? rt method.getReturnType();// 获取参数列表Class?[] pts method.getParameterTypes();// 生成 Object[] args new Object[1...N]StringBuilder code new StringBuilder(Object[] args new Object[).append(pts.length).append(];);for (int j 0; j pts.length; j)// 生成 args[1...N] ($w)$1...N;code.append( args[).append(j).append(] ($w)$).append(j 1).append(;);// 生成 InvokerHandler 接口的 invoker 方法调用语句如下// Object ret handler.invoke(this, methods[1...N], args);code.append( Object ret handler.invoke(this, methods[ ix ], args););// 返回值不为 voidif (!Void.TYPE.equals(rt))// 生成返回语句形如 return (java.lang.String) ret;code.append( return ).append(asArgument(rt, ret)).append(;);methods.add(method);// 添加方法名、访问控制符、参数列表、方法代码等信息到 ClassGenerator 中 ccp.addMethod(method.getName(), method.getModifiers(), rt, pts, method.getExceptionTypes(), code.toString());}}if (pkg null)pkg PACKAGE_NAME;// 构建接口代理类名称pkg .proxy id比如 com.tianxiaobo.proxy0String pcn pkg .proxy id;ccp.setClassName(pcn);ccp.addField(public static java.lang.reflect.Method[] methods;);// 生成 private java.lang.reflect.InvocationHandler handler;ccp.addField(private InvocationHandler.class.getName() handler;);// 为接口代理类添加带有 InvocationHandler 参数的构造方法比如// porxy0(java.lang.reflect.InvocationHandler arg0) {// handler$1;// }ccp.addConstructor(Modifier.PUBLIC, new Class?[]{InvocationHandler.class}, new Class?[0], handler$1;);// 为接口代理类添加默认构造方法ccp.addDefaultConstructor();// 生成接口代理类Class? clazz ccp.toClass();clazz.getField(methods).set(null, methods.toArray(new Method[0]));// 构建 Proxy 子类名称比如 Proxy1Proxy2 等String fcn Proxy.class.getName() id;ccm ClassGenerator.newInstance(cl);ccm.setClassName(fcn);ccm.addDefaultConstructor();ccm.setSuperClass(Proxy.class);// 为 Proxy 的抽象方法 newInstance 生成实现代码形如// public Object newInstance(java.lang.reflect.InvocationHandler h) { // return new com.tianxiaobo.proxy0($1);// }ccm.addMethod(public Object newInstance( InvocationHandler.class.getName() h){ return new pcn ($1); });// 生成 Proxy 实现类Class? pc ccm.toClass();// 通过反射创建 Proxy 实例proxy (Proxy) pc.newInstance();} catch (RuntimeException e) {throw e;} catch (Exception e) {throw new RuntimeException(e.getMessage(), e);} finally {if (ccp ! null)// 释放资源ccp.release();if (ccm ! null)ccm.release();synchronized (cache) {if (proxy null)cache.remove(key);else// 写缓存cache.put(key, new WeakReferenceProxy(proxy));// 唤醒其他等待线程cache.notifyAll();}}return proxy;
} 上面代码比较复杂我也写了很多注释。大家在阅读这段代码时要搞清楚 ccp 和 ccm 的用途不然会被搞晕。ccp 用于为服务接口生成代理类比如我们有一个 DemoService 接口这个接口代理类就是由 ccp 生成的。ccm 则是用于为 org.apache.dubbo.common.bytecode.Proxy 抽象类生成子类主要是实现 Proxy 的抽象方法。下面以 org.apache.dubbo.demo.DemoService 这个接口为例来看一下该接口代理类代码大致是怎样的忽略 EchoService 接口。 package org.apache.dubbo.common.bytecode;public class proxy0 implements org.apache.dubbo.demo.DemoService {public static java.lang.reflect.Method[] methods;private java.lang.reflect.InvocationHandler handler;public proxy0() {}public proxy0(java.lang.reflect.InvocationHandler arg0) {handler $1;}public java.lang.String sayHello(java.lang.String arg0) {Object[] args new Object[1];args[0] ($w) $1;Object ret handler.invoke(this, methods[0], args);return (java.lang.String) ret;}
} 好了到这里代理类生成逻辑就分析完了。整个过程比较复杂大家需要耐心看一下本节点到这里。 4.总结 本篇文章对服务引用的过程进行了较为详尽的分析之所以说是较为详尽是因为还有一些地方没有分析到。比如 Directory、Cluster 等实现类的代码并未进行详细分析由于这些类功能比较独立因此我打算后续单独成文进行分析。暂时我们可以先把这些类看成黑盒只要知道这些类的用途即可。引用服务过程涉及到的调用也非常多大家在阅读相关代码的中耐心些并多进行调试。 好了本篇文章就先到这里了。谢谢阅读。 本文在知识共享许可协议 4.0 下发布转载需在明显位置处注明出处作者田小波本文同步发布在我的个人博客http://www.tianxiaobo.com 本作品采用知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议进行许可。
