dubbo解析-详解元数据中心MetadataReport

1 2 3

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

//simplified参数在入参registryUrl里面 //可以简单认为入参registryUrl包含了几乎所有在客户端配置的参数 //入参consumerUrl只包含将要注册到注册中心的数据 public URL getRegisteredConsumerUrl(final URL consumerUrl, URL registryUrl) { //如果simplified=false，则在consumerUrl基础上添加另外两个配置信息 if (!registryUrl.getParameter(SIMPLIFIED_KEY, false)) { return consumerUrl.addParameters(CATEGORY_KEY, CONSUMERS_CATEGORY, CHECK_KEY, String.valueOf(false)); } else { 如果simplified=true，则从consumerUrl中筛选出5个值，常量定义见下方 return URL.valueOf(consumerUrl, DEFAULT_REGISTER_CONSUMER_KEYS, null).addParameters( CATEGORY_KEY, CONSUMERS_CATEGORY, CHECK_KEY, String.valueOf(false)); } } //DEFAULT_REGISTER_CONSUMER_KEYS定义了简化后注册中心必须有的数据，包括： //1、应用名，key为application //2、应用版本，key为version //3、分组信息，key为group //4、dubbo rpc协议版本，key为dubbo //5、dubbo版本，key为release public static final String[] DEFAULT_REGISTER_CONSUMER_KEYS = { APPLICATION_KEY, VERSION_KEY, GROUP_KEY, DUBBO_VERSION_KEY, RELEASE_KEY };

1 2

consumer://192.168.56.1/com.dubbo.WebsiteProcessor? application=consumer&category=consumers&check=false&dubbo=2.0.2&release=2.7.5

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

if (!registryUrl.getParameter(SIMPLIFIED_KEY, false)) { return providerUrl.removeParameters(getFilteredKeys(providerUrl)).removeParameters( MONITOR_KEY, BIND_IP_KEY, BIND_PORT_KEY, QOS_ENABLE, QOS_HOST, QOS_PORT, ACCEPT_FOREIGN_IP, VALIDATION_KEY, INTERFACES); } else { //下面处理simplified=false //extraKeys为配置dubbo.registry.extra-keys的值 String extraKeys = registryUrl.getParameter(EXTRA_KEYS_KEY, “”); if (!providerUrl.getPath().equals(providerUrl.getParameter(INTERFACE_KEY))) { if (StringUtils.isNotEmpty(extraKeys)) { extraKeys += “,”; } extraKeys += INTERFACE_KEY; } //DEFAULT_REGISTER_PROVIDER_KEYS为要注册到注册中心的数据的key，值见下方 String[] paramsToRegistry = getParamsToRegistry(DEFAULT_REGISTER_PROVIDER_KEYS , COMMA_SPLIT_PATTERN.split(extraKeys)); //组装URL return URL.valueOf(providerUrl, paramsToRegistry, providerUrl.getParameter(METHODS_KEY, (String[]) null)); } public static final String[] DEFAULT_REGISTER_PROVIDER_KEYS = { APPLICATION_KEY, CODEC_KEY, EXCHANGER_KEY, SERIALIZATION_KEY, CLUSTER_KEY, CONNECTIONS_KEY, DEPRECATED_KEY, GROUP_KEY, LOADBALANCE_KEY, MOCK_KEY, PATH_KEY, TIMEOUT_KEY, TOKEN_KEY, VERSION_KEY, WARMUP_KEY, WEIGHT_KEY, TIMESTAMP_KEY, DUBBO_VERSION_KEY, RELEASE_KEY };

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

public AbstractMetadataReport(URL reportServerURL) { setUrl(reportServerURL); //构建本地文件名，APPLICATION_KEY默认不存在，需要在 //MetadataReportConfig.parameters中配置 String defaultFilename = System.getProperty(“user.home”) + “/.dubbo/dubbo-metadata-” + reportServerURL.getParameter(APPLICATION_KEY) + “-” + reportServerURL.getAddress().replaceAll(“:”, “-”) + “.cache”; //如果在MetadataReportConfig.parameters中配置了file参数，则使用它作为文件名 String filename = reportServerURL.getParameter(FILE_KEY, defaultFilename); File file = null; if (ConfigUtils.isNotEmpty(filename)) { file = new File(filename); if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) { if (!file.getParentFile().mkdirs()) { throw new IllegalArgumentException("Invalid service store file " + file + ", cause: Failed to create directory " + file.getParentFile() + “!”); } } if (!initialized.getAndSet(true) && file.exists()) { file.delete();//第一次启动的时候将文件删除 } } this.file = file; loadProperties();//这里调用该方法没有用，因为文件不存在 //设置是否异步将数据传输到元数据中心 syncReport = reportServerURL.getParameter(SYNC_REPORT_KEY, false); //当访问元数据中心失败时，MetadataReportRetry设置了两个与重试相关的参数： //1、重试次数；2、多长时间重试一次 metadataReportRetry = new MetadataReportRetry(reportServerURL.getParameter(RETRY_TIMES_KEY, DEFAULT_METADATA_REPORT_RETRY_TIMES), reportServerURL.getParameter(RETRY_PERIOD_KEY, DEFAULT_METADATA_REPORT_RETRY_PERIOD)); //参数cycle.report表示是否按照一定的频率将元数据更新到元数据中心，默认是true if (reportServerURL.getParameter(CYCLE_REPORT_KEY, DEFAULT_METADATA_REPORT_CYCLE_REPORT)) { //使用定时器按照一定的频率将元数据更新到元数据中心 ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory(“DubboMetadataReportTimer”, true)); //定时任务是每天运行一次，在每天的2:00至6:00一个随机时间 scheduler.scheduleAtFixedRate(this::publishAll, calculateStartTime(), ONE_DAY_IN_MIll, TimeUnit.MILLISECONDS); } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

void publishAll() { logger.info(“start to publish all metadata.”); //allMetadataReports是一个Map<MetadataIdentifier, Object>类型， //存储了客户端和服务端的所有与服务相关的元数据 this.doHandleMetadataCollection(allMetadataReports); } //遍历allMetadataReports中的值， //如果是服务端的元数据，调用storeProviderMetadata； //如果是客户端的元数据，调用storeConsumerMetadata private boolean doHandleMetadataCollection(Map<MetadataIdentifier, Object> metadataMap) { if (metadataMap.isEmpty()) { return true; } Iterator<Map.Entry<MetadataIdentifier, Object>> iterable = metadataMap.entrySet().iterator(); while (iterable.hasNext()) { Map.Entry<MetadataIdentifier, Object> item = iterable.next(); if (PROVIDER_SIDE.equals(item.getKey().getSide())) { this.storeProviderMetadata(item.getKey(), (FullServiceDefinition) item.getValue()); } else if (CONSUMER_SIDE.equals(item.getKey().getSide())) { this.storeConsumerMetadata(item.getKey(), (Map) item.getValue()); } } return false; }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30

public void storeProviderMetadata(MetadataIdentifier providerMetadataIdentifier, ServiceDefinition serviceDefinition) { //是同步保存，还是异步保存，如果异步的话，使用线程池， //reportCacheExecutor是单线程的线程池， //reportCacheExecutor=Executors.newFixedThreadPool(1, new NamedThreadFactory(“DubboSaveMetadataReport”, true)); if (syncReport) { storeProviderMetadataTask(providerMetadataIdentifier, serviceDefinition); } else { reportCacheExecutor.execute(() -> storeProviderMetadataTask(providerMetadataIdentifier, serviceDefinition)); } } private void storeProviderMetadataTask(MetadataIdentifier providerMetadataIdentifier, ServiceDefinition serviceDefinition) { try {//代码有删减 allMetadataReports.put(providerMetadataIdentifier, serviceDefinition); //failedReports记录保存失败的数据，如果保存成功，则删除 failedReports.remove(providerMetadataIdentifier); Gson gson = new Gson(); String data = gson.toJson(serviceDefinition); //doStoreProviderMetadata由子类实现， //例如，ZookeeperMetadataReport是将数据直接保存到zookeeper //providerMetadataIdentifier用于构造zk的路径，data是zk节点上的数据 doStoreProviderMetadata(providerMetadataIdentifier, data); //将数据保存到本地文件，本地文件名在构造方法中已经创建 saveProperties(providerMetadataIdentifier, data, true, !syncReport); } catch (Exception e) { //记录保存失败的元数据 failedReports.put(providerMetadataIdentifier, serviceDefinition); //重试，下面小节分析 metadataReportRetry.startRetryTask(); } }

1 2 3

public boolean retry() { return doHandleMetadataCollection(failedReports); }

1 2 3 4 5 6 7

void cancelRetryTask() { retryScheduledFuture.cancel(false);//取消正在运行的任务 //关闭线程池，之后重试机制无法使用， //当超过重试次数后，dubbo认为元数据中心出现问题，可能宕机或者网络断了 //这有一个缺点，即使元数据中心可以使用了或者网络连通了，重试功能也无法使用了，只能重启应用程序 retryExecutor.shutdown(); }