说明
在api配置，注解配置 可以发现服务订阅是调用RefreceBean的get方法
类图
ReferenceConfig
<1>get
com.alibaba.dubbo.config.ReferenceConfig#get
public synchronized T get() { //是否已经销毁 调用destroy方法会标记为true if (destroyed) { throw new IllegalStateException("Already destroyed!"); } //是否已经订阅 并生成代理类 避免重复定义 if (ref == null) { //<2> init(); } return ref; }
<2>init
com.alibaba.dubbo.config.ReferenceConfig#get
->
com.alibaba.dubbo.config.ReferenceConfig#init
private void init() { //是否已经出初始化 if (initialized) { return; } //标记为初始化 initialized = true; //是否有配置interfaceName if (interfaceName == null || interfaceName.length() == 0) { throw new IllegalStateException("<dubbo:reference interface="" /> interface not allow null!"); } // 检查是否配置consumer 同时配置参数 checkDefault(); //Reference配置参数 //可参考 都是继承同一个父类父类https://www.cnblogs.com/LQBlog/p/12469007.html#autoid-6-5-0 appendProperties(this); //是否是泛型化配置 if (getGeneric() == null && getConsumer() != null) { setGeneric(getConsumer().getGeneric()); } //是否是泛型化接口 if (ProtocolUtils.isGeneric(getGeneric())) { interfaceClass = GenericService.class; } else { try { /** * 反射获得 interface属性配置的class */ interfaceClass = Class.forName(interfaceName, true, Thread.currentThread() .getContextClassLoader()); } catch (ClassNotFoundException e) { throw new IllegalStateException(e.getMessage(), e); } //这里主要是为了检查 如果配置了methods 方法 在当前config是否存在set方法 具体可看https://www.cnblogs.com/LQBlog/p/12469007.html#autoid-6-2-0 checkInterfaceAndMethods(interfaceClass, methods); } /** * 设置当前服务访问地址 从系统变量获取 * -Dcom.test.UserServiceBo=dubbo://30.8.59.182:20880 * 优先级最高 将会忽略zk发现 */ 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) { /** * 也可以通过dubbo.resolve.file 指定配置的properties地址 进行批量设置 * com.test.UserServiceBo=dubbo://30.8.59.182:20880 * com.test.StudentServiceBo=dubbo://30.8.59.182:20880 */ 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 " + resolveFile + ", cause: " + e.getMessage(), e); } 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) { url = resolve; if (logger.isWarnEnabled()) { if (resolveFile != null) { logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service."); } else { logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service."); } } } /** * 以前都是根据优先级获取对应配置 */ if (consumer != null) { if (application == null) { 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(); //检查stup接口配置 checkStub(interfaceClass); //检查mock接口配置 checkMock(interfaceClass); //封装参数 Map<String, String> map = new HashMap<String, String>(); Map<Object, Object> attributes = new HashMap<Object, Object>(); 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()) { //或version String revision = Version.getVersion(interfaceClass, version); //参数追加version if (revision != null && revision.length() > 0) { map.put("revision", revision); } //反射获取方法名字 String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames(); if (methods.length == 0) { logger.warn("NO method found in service interface " + interfaceClass.getName()); map.put("methods", Constants.ANY_VALUE); } else { //追加到参数里面 map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ",")); } } map.put(Constants.INTERFACE_KEY, interfaceName); //反射获得所有get 基本类型方法 以及parameter 封装到map appendParameters(map, application); //反射获得所有get 基本类型方法 以及parameter 封装到map appendParameters(map, module); //反射获得所有get 基本类型方法 以及parameter 封装到map 前缀为default appendParameters(map, consumer, Constants.DEFAULT_KEY); //反射获得当给前对象的所有方法 并封装到map appendParameters(map, this); //group/+interfaacename+version String prefix = StringUtils.getServiceKey(map); //如果配置了method 封装到map if (methods != null && !methods.isEmpty()) { for (MethodConfig method : methods) { appendParameters(map, method, method.getName()); String retryKey = method.getName() + ".retry"; if (map.containsKey(retryKey)) { String retryValue = map.remove(retryKey); if ("false".equals(retryValue)) { map.put(method.getName() + ".retries", "0"); } } appendAttributes(attributes, method, prefix + "." + method.getName()); checkAndConvertImplicitConfig(method, map, attributes); } } //获得注册中心地址 key 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:" + Constants.DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry); } map.put(Constants.REGISTER_IP_KEY, hostToRegistry); //attributes are stored by system context. StaticContext.getSystemContext().putAll(attributes); //<3>生成代理对象 ref = createProxy(map); ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods()); //通过ConsumerModel 封装到订阅列表 ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel); }
<3>createProxy
rivate 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) { // if a url is specified, don't do local reference isJvmRefer = false; //根据url配置判断是否是本地调用 } 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为injvm://127.0.0.1 URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map); /** * SPI扩展点 * private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension(); */ invoker = refprotocol.refer(interfaceClass, url); if (logger.isInfoEnabled()) { logger.info("Using injvm service " + interfaceClass.getName()); } } else { //如果我们手动配置了url;隔开 追加到urls列表 if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address. // 拆分地址成数组，使用 ";" 分隔。 String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url); // 循环数组，添加到 `url` 中。 if (us != null && us.length > 0) { for (String u : us) { // 创建 URL 对象 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 { // assemble URL from register center's configuration //获取注册中心地址 具体可以看你https://www.cnblogs.com/LQBlog/p/12469007.html#autoid-6-10-0 List<URL> us = loadRegistries(false); if (us != null && !us.isEmpty()) { for (URL u : us) { //加载Monitor 使用例子https://blog.csdn.net/sunhuaqiang1/article/details/80141651 URL monitorUrl = loadMonitor(u); if (monitorUrl != null) { map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString())); } //添加refer标识 标识是从注册中心地尼公约 urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map))); } } if (urls.isEmpty()) { 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."); } } // // 单 `urls` 时，引用服务，返回 Invoker 对象 if (urls.size() == 1) { /** * Protocol 取得registryProtocol 不过会被代理 具体可以看 https://www.cnblogs.com/LQBlog/p/12470179.html#autoid-2-0-0 * 默认是registry SPI扩展 private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension(); */ invoker = refprotocol.refer(interfaceClass, urls.get(0)); } else { //集群订阅 List<Invoker<?>> invokers = new ArrayList<Invoker<?>>(); URL registryURL = null; // 循环 `urls` ，引用服务，返回 Invoker 对象 for (URL url : urls) { // 引用服务 invokers.add(refprotocol.refer(interfaceClass, url)); // 使用最后一个注册中心的 URL if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { registryURL = url; // use last registry url } } // // 有注册中心 if (registryURL != null) { // registry url is available // 对有注册中心的 Cluster 只用 AvailableCluster // use AvailableCluster only when register's cluster is available URL u = registryURL.addParameterIfAbsent(Constants.CLUSTER_KEY, AvailableCluster.NAME); invoker = cluster.join(new StaticDirectory(u, invokers)); // 无注册中心，全部都是服务直连 } else { // not a registry url invoker = cluster.join(new StaticDirectory(invokers)); } } } //是否配置了启动检查 Boolean c = check; if (c == null && consumer != null) { c = consumer.isCheck(); } if (c == null) { c = true; // default true } if (c && !invoker.isAvailable()) { // make it possible for consumer to retry later if provider is temporarily unavailable initialized = false; throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion()); } if (logger.isInfoEnabled()) { logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl()); } // 创建代理类 return (T) proxyFactory.getProxy(invoker); }