注册中心在 Dubbo 中的作用

服务治理框架可以大致分为 服务通信 和 服务管理 两部分，服务管理可以分为服务注册、服务订阅以及服务发现，服务提供者 Provider 会往注册中心注册服务，而消费者 Consumer 会从注册中心中订阅自己关注的服务，并在关注的服务发生变更时 得到注册中心的通知。Provider、Consumer 以及 Registry 之间的依赖关系 如下图所示。

avatar

dubbo-registry 模块 结构分析

dubbo 的注册中心有多种实现方案，如：zookeeper、redis、multicast 等，本章先看一下 dubbo-registry 模块的核心部分 dubbo-registry-api，具体实现部分放到下章来讲。dubbo-registry 模块 的结构如下图所示。

avatar

Registry 核心组件类图

典型的 接口 -> 抽象类 -> 实现类 的结构设计，如下图所示。

avatar

既然有 Registry 组件，那么按照很多框架的套路，肯定也有一个用于获取 Registry 实例的 RegistryFactory，其中用到了工厂方法模式，不同的工厂类用于获取不同类型的实例。其类图结构如下。

avatar

源码详解

根据上面的类图，我们开始从上往下 详解 dubbo 中对于注册中心的设计以及实现。

RegistryService 接口

RegistryService 是注册中心模块的服务接口，定义了注册、取消注册、订阅、取消订阅以及查询符合条件的已注册数据 等方法。这里统一说明一下 URL，dubbo 是以总线模式来时刻传递和保存配置信息的，配置信息都被放在 URL 上进行传递，随时可以取得相关配置信息，而这里提到了 URL 有别的作用，就是作为类似于节点的作用，首先服务提供者（Provider）启动时需要提供服务，就会向注册中心写下自己的 URL 地址。然后消费者启动时需要去订阅该服务，则会订阅 Provider 注册的地址，并且消费者也会写下自己的 URL。

  1. /**
  2.  * RegistryService. (SPI, Prototype, ThreadSafe)
  3.  *
  4.  * 注册中心服务接口
  5.  */
  6. public interface RegistryService {
  8.     /**
  9.      * 注册数据，比如：提供者地址，消费者地址，路由规则，覆盖规则 等数据。
  10.      * <p>
  11.      * 注册需处理契约：<br>
  12.      * 1. 当URL设置了check=false时，注册失败后不报错，在后台定时重试，否则抛出异常。<br>
  13.      * 2. 当URL设置了dynamic=false参数，则需持久存储，否则，当注册者出现断电等情况异常退出时，需自动删除。<br>
  14.      * 3. 当URL设置了category=routers时，表示分类存储，缺省类别为providers，可按分类部分通知数据。<br>
  15.      * 4. 当注册中心重启，网络抖动，不能丢失数据，包括断线自动删除数据。<br>
  16.      * 5. 允许URI相同但参数不同的URL并存，不能覆盖。<br>
  17.      *
  18.      * @param url 注册信息，不允许为空，如：dubbo://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
  19.      */
  20.     void register(URL url);
  22.     /**
  23.      * 取消注册.
  24.      * <p>
  25.      * 取消注册需处理契约：<br>
  26.      * 1. 如果是dynamic=false的持久存储数据，找不到注册数据，则抛IllegalStateException，否则忽略。<br>
  27.      * 2. 按全URL匹配取消注册。<br>
  28.      *
  29.      * @param url 注册信息，不允许为空，如：dubbo://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
  30.      */
  31.     void unregister(URL url);
  33.     /**
  34.      * 订阅符合条件的已注册数据，当有注册数据变更时自动推送.
  35.      * <p>
  36.      * 订阅需处理契约：<br>
  37.      * 1. 当URL设置了check=false时，订阅失败后不报错，在后台定时重试。<br>
  38.      * 2. 当URL设置了category=routers，只通知指定分类的数据，多个分类用逗号分隔，并允许星号通配，表示订阅所有分类数据。<br>
  39.      * 3. 允许以interface,group,version,classifier作为条件查询，如：interface=com.alibaba.foo.BarService&version=1.0.0<br>
  40.      * 4. 并且查询条件允许星号通配，订阅所有接口的所有分组的所有版本，或：interface=*&group=*&version=*&classifier=*<br>
  41.      * 5. 当注册中心重启，网络抖动，需自动恢复订阅请求。<br>
  42.      * 6. 允许URI相同但参数不同的URL并存，不能覆盖。<br>
  43.      * 7. 必须阻塞订阅过程，等第一次通知完后再返回。<br>
  44.      *
  45.      * @param url      订阅条件，不允许为空，如：consumer://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
  46.      * @param listener 变更事件监听器，不允许为空
  47.      */
  48.     void subscribe(URL url, NotifyListener listener);
  50.     /**
  51.      * 取消订阅.
  52.      * <p>
  53.      * 取消订阅需处理契约：<br>
  54.      * 1. 如果没有订阅，直接忽略。<br>
  55.      * 2. 按全URL匹配取消订阅。<br>
  56.      *
  57.      * @param url      订阅条件，不允许为空，如：consumer://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
  58.      * @param listener 变更事件监听器，不允许为空
  59.      */
  60.     void unsubscribe(URL url, NotifyListener listener);
  62.     /**
  63.      * 查询符合条件的已注册数据，与订阅的推模式相对应，这里为拉模式，只返回一次结果。
  64.      *
  65.      * @param url 查询条件，不允许为空，如：consumer://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
  66.      * @return 已注册信息列表，可能为空，含义同{@link com.alibaba.dubbo.registry.NotifyListener#notify(List<URL>)}的参数。
  67.      * @see com.alibaba.dubbo.registry.NotifyListener#notify(List)
  68.      */
  69.     List<URL> lookup(URL url);
  70. }

Registry 接口

注册中心接口，把节点 Node 以及注册中心服务 RegistryService 的方法整合在了这个接口里面。该接口并没有自己的方法，就是继承了 Node 和 RegistryService 接口。这里的 Node 是节点的接口，里面协定了关于节点的一些操作方法，源码如下。

  1. /**
  2.  * 注册中心接口
  3.  */
  4. public interface Registry extends Node, RegistryService {
  5. }
  7. public interface Node {
  8.     //获得节点地址
  9.     URL getUrl();
  10.     //判断节点是否可用
  11.     boolean isAvailable();
  12.     //销毁节点
  13.     void destroy();
  14. }

AbstractRegistry 抽象类

实现了 Registry 接口的抽象类。为了减轻注册中心的压力，该抽象类把本地 URL 缓存到了 property 文件中，并且实现了注册中心的注册、订阅等方法。

  1. /**
  2.  * 实现了Registry接口的抽象类，实现了如下方法：
  3.  *
  4.  * 1、通用的注册、订阅、查询、通知等方法
  5.  * 2、读取和持久化注册数据到文件，以 properties 格式存储
  6.  */
  7. public abstract class AbstractRegistry implements Registry {
  9.     // URL地址分隔符，用于文件缓存中，服务提供者URL分隔
  10.     private static final char URL_SEPARATOR = ' ';
  11.     // URL地址分隔正则表达式，用于解析文件缓存中服务提供者URL列表
  12.     private static final String URL_SPLIT = "\\s+";
  14.     // Log output
  15.     protected final Logger logger = LoggerFactory.getLogger(getClass());
  16.     /**
  17.      *  本地磁盘缓存。
  18.      *  1. 其中特殊的 key 值 .registies 记录注册中心列表 TODO 8019 芋艿，特殊的 key 是
  19.      *  2. 其它均为 {@link #notified} 服务提供者列表
  20.      */
  21.     private final Properties properties = new Properties();
  22.     /**
  23.      * 注册中心缓存写入执行器。
  24.      * 线程数=1
  25.      */
  26.     // File cache timing writing
  27.     private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));
  28.     /**
  29.      * 是否同步保存文件
  30.      */
  31.     private final boolean syncSaveFile;
  32.     /**
  33.      * 数据版本号
  34.      */
  35.     private final AtomicLong lastCacheChanged = new AtomicLong();
  36.     /**
  37.      * 已注册 URL 集合。
  38.      * 注册的 URL 可以是服务提供者的，也可以是服务消费者的
  39.      */
  40.     private final Set<URL> registered = new ConcurrentHashSet<URL>();
  41.     /**
  42.      * 订阅 URL 的监听器集合
  43.      * key：订阅者的 URL ，例如消费者的 URL
  44.      */
  45.     private final ConcurrentMap<URL, Set<NotifyListener>> subscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
  46.     /**
  47.      * 被通知的 URL 集合
  48.      * key1：消费者的 URL ，例如消费者的 URL ，和 {@link #subscribed} 的键一致
  49.      * key2：分类，例如：providers、consumers、routes、configurators。【实际无 consumers ，因为消费者不会去订阅另外的消费者的列表】
  50.      *            在 {@link Constants} 中，以 "_CATEGORY" 结尾
  51.      */
  52.     private final ConcurrentMap<URL, Map<String, List<URL>>> notified = new ConcurrentHashMap<URL, Map<String, List<URL>>>();
  53.     /**
  54.      * 注册中心 URL
  55.      */
  56.     private URL registryUrl;
  57.     /**
  58.      * 本地磁盘缓存文件，缓存注册中心的数据
  59.      */
  60.     private File file;
  61.     /**
  62.      * 是否销毁
  63.      */
  64.     private AtomicBoolean destroyed = new AtomicBoolean(false);
  66.     public AbstractRegistry(URL url) {
  67.         setUrl(url);
  68.         // Start file save timer
  69.         syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
  70.         // 获得 `file`
  71.         String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getParameter(Constants.APPLICATION_KEY) + "-" + url.getAddress() + ".cache");
  72.         File file = null;
  73.         if (ConfigUtils.isNotEmpty(filename)) {
  74.             file = new File(filename);
  75.             if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) {
  76.                 if (!file.getParentFile().mkdirs()) {
  77.                     throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
  78.                 }
  79.             }
  80.         }
  81.         this.file = file;
  82.         // 加载本地磁盘缓存文件到内存缓存
  83.         loadProperties();
  84.         // 通知监听器，URL 变化结果
  85.         notify(url.getBackupUrls()); // 【TODO 8020】为什么构造方法，要通知，连监听器都没注册
  86.     }
  88.     protected static List<URL> filterEmpty(URL url, List<URL> urls) {
  89.         if (urls == null || urls.isEmpty()) {
  90.             List<URL> result = new ArrayList<URL>(1);
  91.             result.add(url.setProtocol(Constants.EMPTY_PROTOCOL));
  92.             return result;
  93.         }
  94.         return urls;
  95.     }
  97.     @Override
  98.     public URL getUrl() {
  99.         return registryUrl;
  100.     }
  102.     protected void setUrl(URL url) {
  103.         if (url == null) {
  104.             throw new IllegalArgumentException("registry url == null");
  105.         }
  106.         this.registryUrl = url;
  107.     }
  109.     public Set<URL> getRegistered() {
  110.         return registered;
  111.     }
  113.     public Map<URL, Set<NotifyListener>> getSubscribed() {
  114.         return subscribed;
  115.     }
  117.     public Map<URL, Map<String, List<URL>>> getNotified() {
  118.         return notified;
  119.     }
  121.     public File getCacheFile() {
  122.         return file;
  123.     }
  125.     public Properties getCacheProperties() {
  126.         return properties;
  127.     }
  129.     public AtomicLong getLastCacheChanged() {
  130.         return lastCacheChanged;
  131.     }
  133.     /**
  134.      * 保存内存缓存到本地磁盘缓存文件，即 {@link #properties} => {@link #file}
  135.      *
  136.      * @param version 数据版本号
  137.      */
  138.     public void doSaveProperties(long version) {
  139.         if (version < lastCacheChanged.get()) {
  140.             return;
  141.         }
  142.         if (file == null) {
  143.             return;
  144.         }
  145.         // Save
  146.         try {
  147.             // 创建 .lock 文件
  148.             File lockfile = new File(file.getAbsolutePath() + ".lock");
  149.             if (!lockfile.exists()) {
  150.                 lockfile.createNewFile();
  151.             }
  152.             // 随机读写文件操作
  153.             RandomAccessFile raf = new RandomAccessFile(lockfile, "rw");
  154.             try {
  155.                 FileChannel channel = raf.getChannel();
  156.                 try {
  157.                     // 获得文件锁
  158.                     FileLock lock = channel.tryLock();
  159.                     // 获取失败
  160.                     if (lock == null) {
  161.                         throw new IOException("Can not lock the registry cache file " + file.getAbsolutePath() + ", ignore and retry later, maybe multi java process use the file, please config: dubbo.registry.file=xxx.properties");
  162.                     }
  163.                     // 获取成功，进行保存
  164.                     // Save
  165.                     try {
  166.                         if (!file.exists()) {
  167.                             file.createNewFile();
  168.                         }
  169.                         FileOutputStream outputFile = new FileOutputStream(file);
  170.                         try {
  171.                             properties.store(outputFile, "Dubbo Registry Cache");
  172.                         } finally {
  173.                             outputFile.close();
  174.                         }
  175.                     // 释放文件锁
  176.                     } finally {
  177.                         lock.release();
  178.                     }
  179.                 // 释放文件 Channel
  180.                 } finally {
  181.                     channel.close();
  182.                 }
  183.             // 释放随机读写文件操作
  184.             } finally {
  185.                 raf.close();
  186.             }
  187.         } catch (Throwable e) {
  188.             // 版本号过小，不保存
  189.             if (version < lastCacheChanged.get()) {
  190.                 return;
  191.             // 重新异步保存，一般情况下为上面的获取锁失败抛出的异常。通过这样的方式，达到保存成功。
  192.             } else {
  193.                 registryCacheExecutor.execute(new SaveProperties(lastCacheChanged.incrementAndGet()));
  194.             }
  195.             logger.warn("Failed to save registry store file, cause: " + e.getMessage(), e);
  196.         }
  197.     }
  199.     /**
  200.      * 加载本地磁盘缓存文件到内存缓存，即 {@link #file} => {@link #properties}
  201.      */
  202.     private void loadProperties() {
  203.         if (file != null && file.exists()) {
  204.             InputStream in = null;
  205.             try {
  206.                 // 文件流
  207.                 in = new FileInputStream(file);
  208.                 // 读取文件流
  209.                 properties.load(in);
  210.                 if (logger.isInfoEnabled()) {
  211.                     logger.info("Load registry store file " + file + ", data: " + properties);
  212.                 }
  213.             } catch (Throwable e) {
  214.                 logger.warn("Failed to load registry store file " + file, e);
  215.             } finally {
  216.                 if (in != null) {
  217.                     try {
  218.                         in.close();
  219.                     } catch (IOException e) {
  220.                         logger.warn(e.getMessage(), e);
  221.                     }
  222.                 }
  223.             }
  224.         }
  225.     }
  227.     /**
  228.      * 从 `properties` 中获得缓存的 URL 集合
  229.      *
  230.      * @param url URL
  231.      * @return URL 集合
  232.      */
  233.     public List<URL> getCacheUrls(URL url) {
  234.         for (Map.Entry<Object, Object> entry : properties.entrySet()) {
  235.             String key = (String) entry.getKey();
  236.             String value = (String) entry.getValue();
  237.             if (key != null && key.length() > 0 // 非空
  238.                     && key.equals(url.getServiceKey()) // 服务键匹配
  239.                     && (Character.isLetter(key.charAt(0)) || key.charAt(0) == '_') // TODO 芋艿，_ 是什么
  240.                     && value != null && value.length() > 0) { // 值非空
  241.                 String[] arr = value.trim().split(URL_SPLIT);
  242.                 List<URL> urls = new ArrayList<URL>();
  243.                 for (String u : arr) {
  244.                     urls.add(URL.valueOf(u));
  245.                 }
  246.                 return urls;
  247.             }
  248.         }
  249.         return null;
  250.     }
  252.     @Override
  253.     public List<URL> lookup(URL url) {
  254.         List<URL> result = new ArrayList<URL>();
  255.         Map<String, List<URL>> notifiedUrls = getNotified().get(url);
  256.         // 有数据，遍历数据获得
  257.         if (notifiedUrls != null && notifiedUrls.size() > 0) {
  258.             // 遍历
  259.             for (List<URL> urls : notifiedUrls.values()) {
  260.                 for (URL u : urls) {
  261.                     if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
  262.                         result.add(u);
  263.                     }
  264.                 }
  265.             }
  266.         // 无数据，通过发起订阅的方式得到数据后，遍历数据获得
  267.         } else {
  268.             // 创建 NotifyListener 对象
  269.             final AtomicReference<List<URL>> reference = new AtomicReference<List<URL>>();
  270.             NotifyListener listener = new NotifyListener() {
  271.                 public void notify(List<URL> urls) {
  272.                     reference.set(urls);
  273.                 }
  274.             };
  275.             // 订阅获得数据
  276.             subscribe(url, listener); // Subscribe logic guarantees the first notify to return
  277.             List<URL> urls = reference.get();
  278.             // 遍历
  279.             if (urls != null && !urls.isEmpty()) {
  280.                 for (URL u : urls) {
  281.                     if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
  282.                         result.add(u);
  283.                     }
  284.                 }
  285.             }
  286.         }
  287.         return result;
  288.     }
  290.     @Override
  291.     public void register(URL url) {
  292.         if (url == null) {
  293.             throw new IllegalArgumentException("register url == null");
  294.         }
  295.         if (logger.isInfoEnabled()) {
  296.             logger.info("Register: " + url);
  297.         }
  298.         // 添加到 registered 集合
  299.         registered.add(url);
  300.     }
  302.     @Override
  303.     public void unregister(URL url) {
  304.         if (url == null) {
  305.             throw new IllegalArgumentException("unregister url == null");
  306.         }
  307.         if (logger.isInfoEnabled()) {
  308.             logger.info("Unregister: " + url);
  309.         }
  310.         // 移除出 registered 集合
  311.         registered.remove(url);
  312.     }
  314.     @Override
  315.     public void subscribe(URL url, NotifyListener listener) {
  316.         if (url == null) {
  317.             throw new IllegalArgumentException("subscribe url == null");
  318.         }
  319.         if (listener == null) {
  320.             throw new IllegalArgumentException("subscribe listener == null");
  321.         }
  322.         if (logger.isInfoEnabled()) {
  323.             logger.info("Subscribe: " + url);
  324.         }
  325.         // 添加到 subscribed 集合
  326.         Set<NotifyListener> listeners = subscribed.get(url);
  327.         if (listeners == null) {
  328.             subscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
  329.             listeners = subscribed.get(url);
  330.         }
  331.         listeners.add(listener);
  332.     }
  334.     @Override
  335.     public void unsubscribe(URL url, NotifyListener listener) {
  336.         if (url == null) {
  337.             throw new IllegalArgumentException("unsubscribe url == null");
  338.         }
  339.         if (listener == null) {
  340.             throw new IllegalArgumentException("unsubscribe listener == null");
  341.         }
  342.         if (logger.isInfoEnabled()) {
  343.             logger.info("Unsubscribe: " + url);
  344.         }
  345.         // 移除出 subscribed 集合
  346.         Set<NotifyListener> listeners = subscribed.get(url);
  347.         if (listeners != null) {
  348.             listeners.remove(listener);
  349.         }
  350.     }
  352.     /**
  353.      * 恢复注册和订阅
  354.      *
  355.      * @throws Exception 发生异常
  356.      */
  357.     protected void recover() throws Exception {
  358.         // register 恢复注册
  359.         Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
  360.         if (!recoverRegistered.isEmpty()) {
  361.             if (logger.isInfoEnabled()) {
  362.                 logger.info("Recover register url " + recoverRegistered);
  363.             }
  364.             for (URL url : recoverRegistered) {
  365.                 register(url);
  366.             }
  367.         }
  368.         // subscribe 恢复订阅
  369.         Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
  370.         if (!recoverSubscribed.isEmpty()) {
  371.             if (logger.isInfoEnabled()) {
  372.                 logger.info("Recover subscribe url " + recoverSubscribed.keySet());
  373.             }
  374.             for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
  375.                 URL url = entry.getKey();
  376.                 for (NotifyListener listener : entry.getValue()) {
  377.                     subscribe(url, listener);
  378.                 }
  379.             }
  380.         }
  381.     }
  383.     /**
  384.      * 通知监听器，URL 变化结果。
  385.      *
  386.      * @param urls 通知的 URL 变化结果（全量数据）
  387.      */
  388.     protected void notify(List<URL> urls) {
  389.         if (urls == null || urls.isEmpty()) return;
  390.         // 循环 `subscribed` ，通知监听器们
  391.         for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
  392.             URL url = entry.getKey();
  393.             // 匹配
  394.             if (!UrlUtils.isMatch(url, urls.get(0))) {
  395.                 continue;
  396.             }
  397.             // 通知监听器
  398.             Set<NotifyListener> listeners = entry.getValue();
  399.             if (listeners != null) {
  400.                 for (NotifyListener listener : listeners) {
  401.                     try {
  402.                         notify(url, listener, filterEmpty(url, urls));
  403.                     } catch (Throwable t) {
  404.                         logger.error("Failed to notify registry event, urls: " + urls + ", cause: " + t.getMessage(), t);
  405.                     }
  406.                 }
  407.             }
  408.         }
  409.     }
  411.     /**
  412.      * 通知监听器，URL 变化结果。
  413.      *
  414.      * 数据流向 `urls` => {@link #notified} => {@link #properties} => {@link #file}
  415.      *
  416.      * @param url 消费者 URL
  417.      * @param listener 监听器
  418.      * @param urls 通知的 URL 变化结果（全量数据）
  419.      */
  420.     protected void notify(URL url, NotifyListener listener, List<URL> urls) {
  421.         if (url == null) {
  422.             throw new IllegalArgumentException("notify url == null");
  423.         }
  424.         if (listener == null) {
  425.             throw new IllegalArgumentException("notify listener == null");
  426.         }
  427.         if ((urls == null || urls.isEmpty())
  428.                 && !Constants.ANY_VALUE.equals(url.getServiceInterface())) {
  429.             logger.warn("Ignore empty notify urls for subscribe url " + url);
  430.             return;
  431.         }
  432.         if (logger.isInfoEnabled()) {
  433.             logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
  434.         }
  435.         // 将 `urls` 按照 `url.parameter.category` 分类，添加到集合
  436.         // 注意，特殊情况，使用 curator 连接 Zookeeper 时，若是服务消费者，连接断开，会出现 category=providers,configurations,routes
  437.         Map<String, List<URL>> result = new HashMap<String, List<URL>>();
  438.         for (URL u : urls) {
  439.             if (UrlUtils.isMatch(url, u)) {
  440.                 String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
  441.                 List<URL> categoryList = result.get(category);
  442.                 if (categoryList == null) {
  443.                     categoryList = new ArrayList<URL>();
  444.                     result.put(category, categoryList);
  445.                 }
  446.                 categoryList.add(u);
  447.             }
  448.         }
  449.         if (result.size() == 0) {
  450.             return;
  451.         }
  452.         // 获得消费者 URL 对应的在 `notified` 中，通知的 URL 变化结果（全量数据）
  453.         Map<String, List<URL>> categoryNotified = notified.get(url);
  454.         if (categoryNotified == null) {
  455.             notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
  456.             categoryNotified = notified.get(url);
  457.         }
  458.         // 【按照分类循环】处理通知的 URL 变化结果（全量数据）
  459.         for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
  460.             String category = entry.getKey();
  461.             List<URL> categoryList = entry.getValue();
  462.             // 覆盖到 `notified`
  463.             // 当某个分类的数据为空时，会依然有 urls 。其中 `urls[0].protocol = empty` ，通过这样的方式，处理所有服务提供者为空的情况。
  464.             categoryNotified.put(category, categoryList);
  465.             // 保存到文件
  466.             saveProperties(url);
  467.             // 通知监听器
  468.             listener.notify(categoryList);
  469.         }
  470.     }
  472.     /**
  473.      * 保存单个消费者 URL 对应，在 `notified` 的数据，到文件。
  474.      *
  475.      * @param url 消费者 URL
  476.      */
  477.     private void saveProperties(URL url) {
  478.         if (file == null) {
  479.             return;
  480.         }
  482.         try {
  483.             // 拼接 URL
  484.             StringBuilder buf = new StringBuilder();
  485.             Map<String, List<URL>> categoryNotified = notified.get(url);
  486.             if (categoryNotified != null) {
  487.                 for (List<URL> us : categoryNotified.values()) {
  488.                     for (URL u : us) {
  489.                         if (buf.length() > 0) {
  490.                             buf.append(URL_SEPARATOR);
  491.                         }
  492.                         buf.append(u.toFullString());
  493.                     }
  494.                 }
  495.             }
  496.             // 设置到 properties 中
  497.             properties.setProperty(url.getServiceKey(), buf.toString());
  498.             // 增加数据版本号
  499.             long version = lastCacheChanged.incrementAndGet();
  500.             // 保存到文件
  501.             if (syncSaveFile) {
  502.                 doSaveProperties(version);
  503.             } else {
  504.                 registryCacheExecutor.execute(new SaveProperties(version));
  505.             }
  506.         } catch (Throwable t) {
  507.             logger.warn(t.getMessage(), t);
  508.         }
  509.     }
  511.     /**
  512.      * 取消注册和订阅
  513.      */
  514.     @Override
  515.     public void destroy() {
  516.         // 已销毁，跳过
  517.         if (!destroyed.compareAndSet(false, true)) {
  518.             return;
  519.         }
  520.         if (logger.isInfoEnabled()) {
  521.             logger.info("Destroy registry:" + getUrl());
  522.         }
  523.         // 取消注册
  524.         Set<URL> destroyRegistered = new HashSet<URL>(getRegistered());
  525.         if (!destroyRegistered.isEmpty()) {
  526.             for (URL url : new HashSet<URL>(getRegistered())) {
  527.                 if (url.getParameter(Constants.DYNAMIC_KEY, true)) {
  528.                     try {
  529.                         unregister(url); // 取消注册
  530.                         if (logger.isInfoEnabled()) {
  531.                             logger.info("Destroy unregister url " + url);
  532.                         }
  533.                     } catch (Throwable t) {
  534.                         logger.warn("Failed to unregister url " + url + " to registry " + getUrl() + " on destroy, cause: " + t.getMessage(), t);
  535.                     }
  536.                 }
  537.             }
  538.         }
  539.         // 取消订阅
  540.         Map<URL, Set<NotifyListener>> destroySubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
  541.         if (!destroySubscribed.isEmpty()) {
  542.             for (Map.Entry<URL, Set<NotifyListener>> entry : destroySubscribed.entrySet()) {
  543.                 URL url = entry.getKey();
  544.                 for (NotifyListener listener : entry.getValue()) {
  545.                     try {
  546.                         unsubscribe(url, listener); // 取消订阅
  547.                         if (logger.isInfoEnabled()) {
  548.                             logger.info("Destroy unsubscribe url " + url);
  549.                         }
  550.                     } catch (Throwable t) {
  551.                         logger.warn("Failed to unsubscribe url " + url + " to registry " + getUrl() + " on destroy, cause: " + t.getMessage(), t);
  552.                     }
  553.                 }
  554.             }
  555.         }
  556.     }
  558.     public String toString() {
  559.         return getUrl().toString();
  560.     }
  562.     /**
  563.      * 保存配置的 Runnable任务
  564.      */
  565.     private class SaveProperties implements Runnable {
  567.         /**
  568.          * 数据版本号
  569.          */
  570.         private long version;
  572.         private SaveProperties(long version) {
  573.             this.version = version;
  574.         }
  576.         public void run() {
  577.             doSaveProperties(version);
  578.         }
  579.     }
  580. }

FailbackRegistry 抽象类

FailbackRegistry 抽象类 继承了上面的 AbstractRegistry，AbstractRegistry 中的注册、订阅等方法，实际上就是一些内存缓存的变化，而真正的注册订阅的实现逻辑在 FailbackRegistry 实现，并且 FailbackRegistry 提供了失败重试的机制。

  1. /**
  2.  * 支持失败重试的 FailbackRegistry抽象类
  3.  */
  4. public abstract class FailbackRegistry extends AbstractRegistry {
  6.     /**
  7.      * 定时任务执行器
  8.      */
  9.     private final ScheduledExecutorService retryExecutor = Executors.
  10.             newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryFailedRetryTimer", true));
  12.     /**
  13.      * 失败重试定时器，定时检查是否有请求失败，如有，无限次重试
  14.      */
  15.     private final ScheduledFuture<?> retryFuture;
  16.     /**
  17.      * 注册失败的 URL 集合
  18.      */
  19.     private final Set<URL> failedRegistered = new ConcurrentHashSet<URL>();
  20.     /**
  21.      * 取消注册失败的 URL 集合
  22.      */
  23.     private final Set<URL> failedUnregistered = new ConcurrentHashSet<URL>();
  24.     /**
  25.      * 订阅失败的监听器集合
  26.      */
  27.     private final ConcurrentMap<URL, Set<NotifyListener>> failedSubscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
  28.     /**
  29.      * 取消订阅失败的监听器集合
  30.      */
  31.     private final ConcurrentMap<URL, Set<NotifyListener>> failedUnsubscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
  32.     /**
  33.      * 通知失败的 URL 集合
  34.      */
  35.     private final ConcurrentMap<URL, Map<NotifyListener, List<URL>>> failedNotified = new ConcurrentHashMap<URL, Map<NotifyListener, List<URL>>>();
  37.     /**
  38.      * 是否销毁
  39.      */
  40.     private AtomicBoolean destroyed = new AtomicBoolean(false);
  42.     public FailbackRegistry(URL url) {
  43.         super(url);
  44.         // 重试频率，单位：毫秒
  45.         int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
  46.         // 创建失败重试定时器
  47.         this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
  48.             public void run() {
  49.                 // Check and connect to the registry
  50.                 try {
  51.                     retry();
  52.                 } catch (Throwable t) { // Defensive fault tolerance
  53.                     logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
  54.                 }
  55.             }
  56.         }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
  57.     }
  59.     public Future<?> getRetryFuture() {
  60.         return retryFuture;
  61.     }
  63.     public Set<URL> getFailedRegistered() {
  64.         return failedRegistered;
  65.     }
  67.     public Set<URL> getFailedUnregistered() {
  68.         return failedUnregistered;
  69.     }
  71.     public Map<URL, Set<NotifyListener>> getFailedSubscribed() {
  72.         return failedSubscribed;
  73.     }
  75.     public Map<URL, Set<NotifyListener>> getFailedUnsubscribed() {
  76.         return failedUnsubscribed;
  77.     }
  79.     public Map<URL, Map<NotifyListener, List<URL>>> getFailedNotified() {
  80.         return failedNotified;
  81.     }
  83.     /**
  84.      * 添加到 `failedSubscribed`
  85.      */
  86.     private void addFailedSubscribed(URL url, NotifyListener listener) {
  87.         Set<NotifyListener> listeners = failedSubscribed.get(url);
  88.         if (listeners == null) {
  89.             failedSubscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
  90.             listeners = failedSubscribed.get(url);
  91.         }
  92.         listeners.add(listener);
  93.     }
  95.     /**
  96.      * 移除出 `failedSubscribed` `failedUnsubscribed` `failedNotified`
  97.      */
  98.     private void removeFailedSubscribed(URL url, NotifyListener listener) {
  99.         // 移除出 `failedSubscribed`
  100.         Set<NotifyListener> listeners = failedSubscribed.get(url);
  101.         if (listeners != null) {
  102.             listeners.remove(listener);
  103.         }
  104.         // 移除出 `failedUnsubscribed`
  105.         listeners = failedUnsubscribed.get(url);
  106.         if (listeners != null) {
  107.             listeners.remove(listener);
  108.         }
  109.         // 移除出 `failedNotified`
  110.         Map<NotifyListener, List<URL>> notified = failedNotified.get(url);
  111.         if (notified != null) {
  112.             notified.remove(listener);
  113.         }
  114.     }
  116.     @Override
  117.     public void register(URL url) {
  118.         // 已销毁，跳过
  119.         if (destroyed.get()){
  120.             return;
  121.         }
  122.         // 添加到 `registered` 变量
  123.         super.register(url);
  124.         // 移除出 `failedRegistered` `failedUnregistered` 变量
  125.         failedRegistered.remove(url);
  126.         failedUnregistered.remove(url);
  127.         // 向注册中心发送注册请求
  128.         try {
  129.             doRegister(url);
  130.         } catch (Exception e) {
  131.             Throwable t = e;
  133.             // 如果开启了启动时检测，则直接抛出异常
  134.             boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
  135.                     && url.getParameter(Constants.CHECK_KEY, true)
  136.                     && !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol()); // 非消费者。消费者会在 `ReferenceConfig#createProxy(...)` 方法中，调用 `Invoker#avalible()` 方法，进行检查。
  137.             boolean skipFailback = t instanceof SkipFailbackWrapperException;
  138.             if (check || skipFailback) {
  139.                 if (skipFailback) {
  140.                     t = t.getCause();
  141.                 }
  142.                 throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
  143.             } else {
  144.                 logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  145.             }
  147.             // 将失败的注册请求记录到 `failedRegistered`，定时重试
  148.             failedRegistered.add(url);
  149.         }
  150.     }
  152.     @Override
  153.     public void unregister(URL url) {
  154.         // 已销毁，跳过
  155.         if (destroyed.get()){
  156.             return;
  157.         }
  158.         // 移除出 `registered` 变量
  159.         super.unregister(url);
  160.         // 移除出 `failedRegistered` `failedUnregistered` 变量
  161.         failedRegistered.remove(url);
  162.         failedUnregistered.remove(url);
  163.         // 向注册中心发送取消注册请求
  164.         try {
  165.             doUnregister(url);
  166.         } catch (Exception e) {
  167.             Throwable t = e;
  169.             // 如果开启了启动时检测，则直接抛出异常
  170.             boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
  171.                     && url.getParameter(Constants.CHECK_KEY, true)
  172.                     && !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
  173.             boolean skipFailback = t instanceof SkipFailbackWrapperException;
  174.             if (check || skipFailback) {
  175.                 if (skipFailback) {
  176.                     t = t.getCause();
  177.                 }
  178.                 throw new IllegalStateException("Failed to unregister " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
  179.             } else {
  180.                 logger.error("Failed to uregister " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  181.             }
  183.             // 将失败的取消注册请求记录到 `failedUnregistered`，定时重试
  184.             failedUnregistered.add(url);
  185.         }
  186.     }
  188.     @Override
  189.     public void subscribe(URL url, NotifyListener listener) {
  190.         // 已销毁，跳过
  191.         if (destroyed.get()){
  192.             return;
  193.         }
  194.         // 移除出 `subscribed` 变量
  195.         super.subscribe(url, listener);
  196.         // 移除出 `failedSubscribed` `failedUnsubscribed` `failedNotified`
  197.         removeFailedSubscribed(url, listener);
  198.         // 向注册中心发送订阅请求
  199.         try {
  200.             doSubscribe(url, listener);
  201.         } catch (Exception e) {
  202.             Throwable t = e;
  204.             // 如果有缓存的 URL 集合，进行通知。后续订阅成功后，会使用最新的 URL 集合，进行通知。
  205.             List<URL> urls = getCacheUrls(url);
  206.             if (urls != null && !urls.isEmpty()) {
  207.                 notify(url, listener, urls);
  208.                 logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
  209.             } else {
  210.                 // 如果开启了启动时检测，则直接抛出异常
  211.                 // If the startup detection is opened, the Exception is thrown directly.
  212.                 boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
  213.                         && url.getParameter(Constants.CHECK_KEY, true);
  214.                 boolean skipFailback = t instanceof SkipFailbackWrapperException;
  215.                 if (check || skipFailback) {
  216.                     if (skipFailback) {
  217.                         t = t.getCause();
  218.                     }
  219.                     throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t);
  220.                 } else {
  221.                     logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  222.                 }
  223.             }
  225.             // 将失败的订阅请求记录到 `failedSubscribed`，定时重试
  226.             // Record a failed registration request to a failed list, retry regularly
  227.             addFailedSubscribed(url, listener);
  228.         }
  229.     }
  231.     @Override
  232.     public void unsubscribe(URL url, NotifyListener listener) {
  233.         // 已销毁，跳过
  234.         if (destroyed.get()){
  235.             return;
  236.         }
  237.         // 移除出 `unsubscribed` 变量
  238.         super.unsubscribe(url, listener);
  239.         // 移除出 `failedSubscribed` `failedUnsubscribed` `failedNotified`
  240.         removeFailedSubscribed(url, listener);
  241.         // 向注册中心发送取消订阅请求
  242.         try {
  243.             // Sending a canceling subscription request to the server side
  244.             doUnsubscribe(url, listener);
  245.         } catch (Exception e) {
  246.             Throwable t = e;
  248.             // 如果开启了启动时检测，则直接抛出异常
  249.             // If the startup detection is opened, the Exception is thrown directly.
  250.             boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
  251.                     && url.getParameter(Constants.CHECK_KEY, true);
  252.             boolean skipFailback = t instanceof SkipFailbackWrapperException;
  253.             if (check || skipFailback) {
  254.                 if (skipFailback) {
  255.                     t = t.getCause();
  256.                 }
  257.                 throw new IllegalStateException("Failed to unsubscribe " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
  258.             } else {
  259.                 logger.error("Failed to unsubscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  260.             }
  262.             // 将失败的订阅请求记录到 `failedUnsubscribed`，定时重试
  263.             // Record a failed registration request to a failed list, retry regularly
  264.             Set<NotifyListener> listeners = failedUnsubscribed.get(url);
  265.             if (listeners == null) {
  266.                 failedUnsubscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
  267.                 listeners = failedUnsubscribed.get(url);
  268.             }
  269.             listeners.add(listener);
  270.         }
  271.     }
  273.     @Override
  274.     protected void notify(URL url, NotifyListener listener, List<URL> urls) {
  275.         if (url == null) {
  276.             throw new IllegalArgumentException("notify url == null");
  277.         }
  278.         if (listener == null) {
  279.             throw new IllegalArgumentException("notify listener == null");
  280.         }
  281.         // 通知监听器
  282.         try {
  283.             doNotify(url, listener, urls);
  284.         } catch (Exception t) {
  285.             // 将失败的通知记录到 `failedNotified`，定时重试
  286.             Map<NotifyListener, List<URL>> listeners = failedNotified.get(url);
  287.             if (listeners == null) {
  288.                 failedNotified.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, List<URL>>());
  289.                 listeners = failedNotified.get(url);
  290.             }
  291.             listeners.put(listener, urls);
  292.             logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  293.         }
  294.     }
  296.     protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
  297.         super.notify(url, listener, urls);
  298.     }
  300.     @Override
  301.     protected void recover() throws Exception {
  302.         // register 恢复注册，添加到 `failedRegistered` ，定时重试
  303.         Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
  304.         if (!recoverRegistered.isEmpty()) {
  305.             if (logger.isInfoEnabled()) {
  306.                 logger.info("Recover register url " + recoverRegistered);
  307.             }
  308.             for (URL url : recoverRegistered) {
  309.                 failedRegistered.add(url);
  310.             }
  311.         }
  312.         // subscribe 恢复订阅，添加到 `failedSubscribed` ，定时重试
  313.         Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
  314.         if (!recoverSubscribed.isEmpty()) {
  315.             if (logger.isInfoEnabled()) {
  316.                 logger.info("Recover subscribe url " + recoverSubscribed.keySet());
  317.             }
  318.             for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
  319.                 URL url = entry.getKey();
  320.                 for (NotifyListener listener : entry.getValue()) {
  321.                     addFailedSubscribed(url, listener);
  322.                 }
  323.             }
  324.         }
  325.     }
  327.     /**
  328.      * 重试
  329.      */
  330.     protected void retry() {
  331.         // 重试执行注册
  332.         if (!failedRegistered.isEmpty()) {
  333.             Set<URL> failed = new HashSet<URL>(failedRegistered); // 避免并发冲突
  334.             if (failed.size() > 0) {
  335.                 if (logger.isInfoEnabled()) {
  336.                     logger.info("Retry register " + failed);
  337.                 }
  338.                 try {
  339.                     for (URL url : failed) {
  340.                         try {
  341.                             // 执行注册
  342.                             doRegister(url);
  343.                             // 移除出 `failedRegistered`
  344.                             failedRegistered.remove(url);
  345.                         } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  346.                             logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  347.                         }
  348.                     }
  349.                 } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  350.                     logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  351.                 }
  352.             }
  353.         }
  354.         // 重试执行取消注册
  355.         if (!failedUnregistered.isEmpty()) {
  356.             Set<URL> failed = new HashSet<URL>(failedUnregistered); // 避免并发冲突
  357.             if (!failed.isEmpty()) {
  358.                 if (logger.isInfoEnabled()) {
  359.                     logger.info("Retry unregister " + failed);
  360.                 }
  361.                 try {
  362.                     for (URL url : failed) {
  363.                         try {
  364.                             // 执行取消注册
  365.                             doUnregister(url);
  366.                             // 移除出 `failedUnregistered`
  367.                             failedUnregistered.remove(url);
  368.                         } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  369.                             logger.warn("Failed to retry unregister  " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  370.                         }
  371.                     }
  372.                 } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  373.                     logger.warn("Failed to retry unregister  " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  374.                 }
  375.             }
  376.         }
  377.         // 重试执行注册
  378.         if (!failedSubscribed.isEmpty()) {
  379.             Map<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedSubscribed); // 避免并发冲突
  380.             for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>(failed).entrySet()) {
  381.                 if (entry.getValue() == null || entry.getValue().size() == 0) {
  382.                     failed.remove(entry.getKey());
  383.                 }
  384.             }
  385.             if (failed.size() > 0) {
  386.                 if (logger.isInfoEnabled()) {
  387.                     logger.info("Retry subscribe " + failed);
  388.                 }
  389.                 try {
  390.                     for (Map.Entry<URL, Set<NotifyListener>> entry : failed.entrySet()) {
  391.                         URL url = entry.getKey();
  392.                         Set<NotifyListener> listeners = entry.getValue();
  393.                         for (NotifyListener listener : listeners) {
  394.                             try {
  395.                                 // 执行注册
  396.                                 doSubscribe(url, listener);
  397.                                 // 移除出监听器
  398.                                 listeners.remove(listener);
  399.                             } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  400.                                 logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  401.                             }
  402.                         }
  403.                     }
  404.                 } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  405.                     logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  406.                 }
  407.             }
  408.         }
  409.         // 重试执行取消注册
  410.         if (!failedUnsubscribed.isEmpty()) {
  411.             Map<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedUnsubscribed);
  412.             for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>(failed).entrySet()) {
  413.                 if (entry.getValue() == null || entry.getValue().isEmpty()) {
  414.                     failed.remove(entry.getKey());
  415.                 }
  416.             }
  417.             if (failed.size() > 0) {
  418.                 if (logger.isInfoEnabled()) {
  419.                     logger.info("Retry unsubscribe " + failed);
  420.                 }
  421.                 try {
  422.                     for (Map.Entry<URL, Set<NotifyListener>> entry : failed.entrySet()) {
  423.                         URL url = entry.getKey();
  424.                         Set<NotifyListener> listeners = entry.getValue();
  425.                         for (NotifyListener listener : listeners) {
  426.                             try {
  427.                                 // 执行取消注册
  428.                                 doUnsubscribe(url, listener);
  429.                                 // 移除出监听器
  430.                                 listeners.remove(listener);
  431.                             } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  432.                                 logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  433.                             }
  434.                         }
  435.                     }
  436.                 } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  437.                     logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  438.                 }
  439.             }
  440.         }
  441.         // 重试执行通知监听器
  442.         if (!failedNotified.isEmpty()) {
  443.             Map<URL, Map<NotifyListener, List<URL>>> failed = new HashMap<URL, Map<NotifyListener, List<URL>>>(failedNotified);
  444.             for (Map.Entry<URL, Map<NotifyListener, List<URL>>> entry : new HashMap<URL, Map<NotifyListener, List<URL>>>(failed).entrySet()) {
  445.                 if (entry.getValue() == null || entry.getValue().size() == 0) {
  446.                     failed.remove(entry.getKey());
  447.                 }
  448.             }
  449.             if (failed.size() > 0) {
  450.                 if (logger.isInfoEnabled()) {
  451.                     logger.info("Retry notify " + failed);
  452.                 }
  453.                 try {
  454.                     for (Map<NotifyListener, List<URL>> values : failed.values()) {
  455.                         for (Map.Entry<NotifyListener, List<URL>> entry : values.entrySet()) {
  456.                             try {
  457.                                 NotifyListener listener = entry.getKey();
  458.                                 List<URL> urls = entry.getValue();
  459.                                 // 通知监听器
  460.                                 listener.notify(urls);
  461.                                 // 移除出监听器
  462.                                 values.remove(listener);
  463.                             } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  464.                                 logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  465.                             }
  466.                         }
  467.                     }
  468.                 } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
  469.                     logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
  470.                 }
  471.             }
  472.         }
  473.     }
  475.     @Override
  476.     public void destroy() {
  477.         // 忽略，若已经销毁
  478.         if (!canDestroy()) {
  479.             return;
  480.         }
  481.         // 调用父方法，取消注册和订阅
  482.         super.destroy();
  483.         // 销毁重试任务
  484.         try {
  485.             retryFuture.cancel(true);
  486.         } catch (Throwable t) {
  487.             logger.warn(t.getMessage(), t);
  488.         }
  489.     }
  491.     // TODO: 2017/8/30 to abstract this method
  492.     protected boolean canDestroy(){
  493.         return destroyed.compareAndSet(false, true);
  494.     }
  496.     // ==== Template method ====
  498.     protected abstract void doRegister(URL url);
  500.     protected abstract void doUnregister(URL url);
  502.     protected abstract void doSubscribe(URL url, NotifyListener listener);
  504.     protected abstract void doUnsubscribe(URL url, NotifyListener listener);
  505. }

RegistryFactory 和 AbstractRegistryFactory

RegistryFactory 接口 是 Registry 的工厂接口，用来返回 Registry 实例。该接口是一个可扩展接口，可以看到该接口上有个@SPI 注解，并且默认值为 dubbo，也就是默认扩展的是 DubboRegistryFactory。AbstractRegistryFactory 则是实现了 RegistryFactory 接口 的抽象类。其源码如下。

  1. /**
  2.  * 注册中心工厂
  3.  */
  4. @SPI("dubbo")
  5. public interface RegistryFactory {
  7.     /**
  8.      * 根据注册中心连接地址，获取注册中心实例
  9.      * <p>
  10.      * 连接注册中心需处理契约：<br>
  11.      * 1. 当设置check=false时表示不检查连接，否则在连接不上时抛出异常。<br>
  12.      * 2. 支持URL上的username:password权限认证。<br>
  13.      * 3. 支持backup=10.20.153.10备选注册中心集群地址。<br>
  14.      * 4. 支持file=registry.cache本地磁盘文件缓存。<br>
  15.      * 5. 支持timeout=1000请求超时设置。<br>
  16.      * 6. 支持session=60000会话超时或过期设置。<br>
  17.      *
  18.      * @param url 注册中心地址，不允许为空
  19.      * @return 注册中心引用，总不返回空
  20.      */
  21.     @Adaptive({"protocol"})
  22.     Registry getRegistry(URL url);
  23. }
  25. /**
  26.  * 注册中心抽象类
  27.  */
  28. public abstract class AbstractRegistryFactory implements RegistryFactory {
  30.     // Log output
  31.     private static final Logger LOGGER = LoggerFactory.getLogger(AbstractRegistryFactory.class);
  33.     // The lock for the acquisition process of the registry
  34.     private static final ReentrantLock LOCK = new ReentrantLock();
  36.     /**
  37.      * Registry 集合
  38.      */
  39.     private static final Map<String, Registry> REGISTRIES = new ConcurrentHashMap<String, Registry>();
  41.     /**
  42.      * Get all registries
  43.      */
  44.     public static Collection<Registry> getRegistries() {
  45.         return Collections.unmodifiableCollection(REGISTRIES.values());
  46.     }
  48.     /**
  49.      * 销毁所有 Registry
  50.      */
  51.     // TODO: 2017/8/30 to move somewhere else better
  52.     public static void destroyAll() {
  53.         if (LOGGER.isInfoEnabled()) {
  54.             LOGGER.info("Close all registries " + getRegistries());
  55.         }
  56.         // 获得锁
  57.         LOCK.lock();
  58.         try {
  59.             // 销毁
  60.             for (Registry registry : getRegistries()) {
  61.                 try {
  62.                     registry.destroy();
  63.                 } catch (Throwable e) {
  64.                     LOGGER.error(e.getMessage(), e);
  65.                 }
  66.             }
  67.             // 清空缓存
  68.             REGISTRIES.clear();
  69.         } finally {
  70.             // 释放锁
  71.             LOCK.unlock();
  72.         }
  73.     }
  75.     /**
  76.      * 获得注册中心 Registry 对象
  77.      *
  78.      * @param url 注册中心地址，不允许为空
  79.      * @return Registry 对象
  80.      */
  81.     @Override
  82.     public Registry getRegistry(URL url) {
  83.         // 修改 URL
  84.         url = url.setPath(RegistryService.class.getName()) // + `path`
  85.                 .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName()) // + `parameters.interface`
  86.                 .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY); // - `export`
  87.         // 计算 key
  88.         String key = url.toServiceString();
  89.         // 获得锁
  90.         // Lock the registry access process to ensure a single instance of the registry
  91.         LOCK.lock();
  92.         try {
  93.             // 从缓存中获得 Registry 对象
  94.             Registry registry = REGISTRIES.get(key);
  95.             if (registry != null) {
  96.                 return registry;
  97.             }
  98.             // 缓存不存在，进行创建 Registry 对象
  99.             registry = createRegistry(url);
  100.             if (registry == null) {
  101.                 throw new IllegalStateException("Can not create registry " + url);
  102.             }
  103.             // 添加到缓存
  104.             REGISTRIES.put(key, registry);
  105.             return registry;
  106.         } finally {
  107.             // 释放锁
  108.             // Release the lock
  109.             LOCK.unlock();
  110.         }
  111.     }
  113.     /**
  114.      * 创建 Registry 对象
  115.      *
  116.      * @param url 注册中心地址
  117.      * @return Registry 对象
  118.      */
  119.     protected abstract Registry createRegistry(URL url);
  120. }

NotifyListener 和 RegistryDirectory

最后我们来看一下 dubbo-registry-api 模块下的另一个比较重要的组件，NotifyListener 接口 和 RegistryDirectory 抽象类。NotifyListener 接口 只有一个 notify 方法，通知监听器。当收到服务变更通知时触发。RegistryDirectory 是注册中心服务，维护着所有可用的远程 Invoker 或者本地的 Invoker，它的 Invoker 集合是从注册中心获取的，另外，它实现了 NotifyListener 接口。比如消费方要调用某远程服务，会向注册中心订阅这个服务的所有 服务提供方，在订阅 及 服务提供方数据有变动时，回调消费方的 NotifyListener 服务的 notify 方法，回调接口传入所有服务提供方的 url 地址然后将 urls 转化为 invokers，也就是 refer 应用远程服务。源码如下。

  1. /**
  2.  * 通知监听器
  3.  */
  4. public interface NotifyListener {
  6.     /**
  7.      * 当收到服务变更通知时触发。
  8.      * <p>
  9.      * 通知需处理契约：<br>
  10.      * 1. 总是以服务接口和数据类型为维度全量通知，即不会通知一个服务的同类型的部分数据，用户不需要对比上一次通知结果。<br>
  11.      * 2. 订阅时的第一次通知，必须是一个服务的所有类型数据的全量通知。<br>
  12.      * 3. 中途变更时，允许不同类型的数据分开通知，比如：providers, consumers, routers, overrides，允许只通知其中一种类型，但该类型的数据必须是全量的，不是增量的。<br>
  13.      * 4. 如果一种类型的数据为空，需通知一个empty协议并带category参数的标识性URL数据。<br>
  14.      * 5. 通知者(即注册中心实现)需保证通知的顺序，比如：单线程推送，队列串行化，带版本对比。<br>
  15.      *
  16.      * @param urls 已注册信息列表，总不为空，含义同{@link com.alibaba.dubbo.registry.RegistryService#lookup(URL)}的返回值。
  17.      */
  18.     void notify(List<URL> urls);
  19. }
  22. /**
  23.  * 基于注册中心的 Directory 实现类
  24.  */
  25. public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
  27.     private static final Logger logger = LoggerFactory.getLogger(RegistryDirectory.class);
  29.     // ========== Dubbo SPI Adaptive 对象 BEGIN ==========
  31.     /**
  32.      * Cluster$Adaptive 对象
  33.      */
  34.     private static final Cluster cluster = ExtensionLoader.getExtensionLoader(Cluster.class).getAdaptiveExtension();
  35.     /**
  36.      * RouterFactory$Adaptive 对象
  37.      */
  38.     private static final RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getAdaptiveExtension();
  39.     /**
  40.      * ConfiguratorFactory$Adaptive 对象
  41.      */
  42.     private static final ConfiguratorFactory configuratorFactory = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class).getAdaptiveExtension();
  44.     // ========== 服务消费者相关 BEGIN ==========
  46.     /**
  47.      * 服务类型，例如：com.alibaba.dubbo.demo.DemoService
  48.      */
  49.     private final Class<T> serviceType; // Initialization at construction time, assertion not null
  50.     /**
  51.      * Consumer URL 的配置项 Map
  52.      */
  53.     private final Map<String, String> queryMap; // Initialization at construction time, assertion not null
  54.     /**
  55.      * 服务方法数组
  56.      */
  57.     private final String[] serviceMethods;
  58.     /**
  59.      * 是否引用多分组
  60.      *
  61.      * 服务分组：https://dubbo.gitbooks.io/dubbo-user-book/demos/service-group.html
  62.      */
  63.     private final boolean multiGroup;
  65.     // ========== 注册中心相关 BEGIN ==========
  67.     /**
  68.      * 注册中心的 Protocol 对象
  69.      */
  70.     private Protocol protocol; // Initialization at the time of injection, the assertion is not null
  71.     /**
  72.      * 注册中心
  73.      */
  74.     private Registry registry; // Initialization at the time of injection, the assertion is not null
  75.     /**
  76.      * 注册中心的服务类，目前是 com.alibaba.dubbo.registry.RegistryService
  77.      *
  78.      * 通过 {@link #url} 的 {@link URL#getServiceKey()} 获得
  79.      */
  80.     private final String serviceKey; // Initialization at construction time, assertion not null
  81.     /**
  82.      * 是否禁止访问。
  83.      *
  84.      * 有两种情况会导致：
  85.      *
  86.      * 1. 没有服务提供者
  87.      * 2. 服务提供者被禁用
  88.      */
  89.     private volatile boolean forbidden = false;
  91.     // ========== 配置规则相关 BEGIN ==========
  93.     /**
  94.      * 原始的目录 URL
  95.      *
  96.      * 例如：zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-consumer&callbacks=1000&check=false&client=netty4&cluster=failback&dubbo=2.0.0&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello,callbackParam,save,update,say03,delete,say04,demo,say01,bye,say02,saves&payload=1000&pid=63400&qos.port=33333&register.ip=192.168.16.23&sayHello.async=true&side=consumer&timeout=10000&timestamp=1527056491064
  97.      */
  98.     private final URL directoryUrl; // Initialization at construction time, assertion not null, and always assign non null value
  99.     /**
  100.      * 覆写的目录 URL ，结合配置规则
  101.      */
  102.     private volatile URL overrideDirectoryUrl; // Initialization at construction time, assertion not null, and always assign non null value
  103.     /**
  104.      * 配置规则数组
  105.      *
  106.      * override rules
  107.      * Priority: override>-D>consumer>provider
  108.      * Rule one: for a certain provider <ip:port,timeout=100>
  109.      * Rule two: for all providers <* ,timeout=5000>
  110.      */
  111.     private volatile List<Configurator> configurators; // The initial value is null and the midway may be assigned to null, please use the local variable reference
  113.     // ========== 服务提供者相关 BEGIN ==========
  115.     /**
  116.      * [url]与[服务提供者 Invoker 集合]的映射缓存
  117.      */
  118.     // Map<url, Invoker> cache service url to invoker mapping.
  119.     private volatile Map<String, Invoker<T>> urlInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference
  120.     /**
  121.      * [方法名]与[服务提供者 Invoker 集合]的映射缓存
  122.      */
  123.     // Map<methodName, Invoker> cache service method to invokers mapping.
  124.     private volatile Map<String, List<Invoker<T>>> methodInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference
  125.     /**
  126.      * [服务提供者 Invoker 集合]缓存
  127.      */
  128.     // Set<invokerUrls> cache invokeUrls to invokers mapping.
  129.     private volatile Set<URL> cachedInvokerUrls; // The initial value is null and the midway may be assigned to null, please use the local variable reference
  131.     public RegistryDirectory(Class<T> serviceType, URL url) {
  132.         super(url);
  133.         if (serviceType == null) {
  134.             throw new IllegalArgumentException("service type is null.");
  135.         }
  136.         if (url.getServiceKey() == null || url.getServiceKey().length() == 0) {
  137.             throw new IllegalArgumentException("registry serviceKey is null.");
  138.         }
  139.         this.serviceType = serviceType;
  140.         this.serviceKey = url.getServiceKey();
  141.         // 获得 queryMap
  142.         this.queryMap = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
  143.         // 获得 overrideDirectoryUrl 和 directoryUrl
  144.         this.overrideDirectoryUrl = this.directoryUrl = url.setPath(url.getServiceInterface()).clearParameters().addParameters(queryMap).removeParameter(Constants.MONITOR_KEY);
  145.         // 初始化 multiGroup
  146.         String group = directoryUrl.getParameter(Constants.GROUP_KEY, "");
  147.         this.multiGroup = group != null && ("*".equals(group) || group.contains(","));
  148.         // 初始化 serviceMethods
  149.         String methods = queryMap.get(Constants.METHODS_KEY);
  150.         this.serviceMethods = methods == null ? null : Constants.COMMA_SPLIT_PATTERN.split(methods);
  151.     }
  153.     /**
  154.      * 将overrideURL 转换为 map，供重新 refer 时使用.
  155.      * 每次下发全部规则，全部重新组装计算
  156.      *
  157.      * @param urls 契约：
  158.      *             </br>1.override://0.0.0.0/...(或override://ip:port...?anyhost=true)&para1=value1...表示全局规则(对所有的提供者全部生效)
  159.      *             </br>2.override://ip:port...?anyhost=false 特例规则（只针对某个提供者生效）
  160.      *             </br>3.不支持override://规则... 需要注册中心自行计算.
  161.      *             </br>4.不带参数的override://0.0.0.0/ 表示清除override
  162.      *
  163.      * @return Configurator 集合
  164.      */
  165.     public static List<Configurator> toConfigurators(List<URL> urls) {
  166.         // 忽略，若配置规则 URL 集合为空
  167.         if (urls == null || urls.isEmpty()) {
  168.             return Collections.emptyList();
  169.         }
  171.         // 创建 Configurator 集合
  172.         List<Configurator> configurators = new ArrayList<Configurator>(urls.size());
  173.         for (URL url : urls) {
  174.             // 若协议为 `empty://` ，意味着清空所有配置规则，因此返回空 Configurator 集合
  175.             if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
  176.                 configurators.clear();
  177.                 break;
  178.             }
  179.             // 对应第 4 条契约，不带参数的 override://0.0.0.0/ 表示清除 override
  180.             Map<String, String> override = new HashMap<String, String>(url.getParameters());
  181.             // The anyhost parameter of override may be added automatically, it can't change the judgement of changing url
  182.             // override 上的 anyhost 可能是自动添加的，不能影响改变url判断
  183.             override.remove(Constants.ANYHOST_KEY);
  184.             if (override.size() == 0) {
  185.                 configurators.clear();
  186.                 continue;
  187.             }
  188.             // 获得 Configurator 对象，并添加到 `configurators` 中
  189.             configurators.add(configuratorFactory.getConfigurator(url));
  190.         }
  191.         // 排序
  192.         Collections.sort(configurators);
  193.         return configurators;
  194.     }
  196.     public void setProtocol(Protocol protocol) {
  197.         this.protocol = protocol;
  198.     }
  200.     public void setRegistry(Registry registry) {
  201.         this.registry = registry;
  202.     }
  204.     /**
  205.      * 发起订阅
  206.      *
  207.      * @param url 消费者 URL
  208.      */
  209.     public void subscribe(URL url) {
  210.         // 设置消费者 URL
  211.         setConsumerUrl(url);
  212.         // 向注册中心，发起订阅
  213.         registry.subscribe(url, this);
  214.     }
  216.     @Override
  217.     public void destroy() {
  218.         if (isDestroyed()) {
  219.             return;
  220.         }
  221.         // 取消订阅
  222.         // unsubscribe.
  223.         try {
  224.             if (getConsumerUrl() != null && registry != null && registry.isAvailable()) {
  225.                 registry.unsubscribe(getConsumerUrl(), this);
  226.             }
  227.         } catch (Throwable t) {
  228.             logger.warn("unexpeced error when unsubscribe service " + serviceKey + "from registry" + registry.getUrl(), t);
  229.         }
  230.         // 标记已经销毁
  231.         super.destroy(); // must be executed after unsubscribing
  232.         // 销毁所有 Invoker
  233.         try {
  234.             destroyAllInvokers();
  235.         } catch (Throwable t) {
  236.             logger.warn("Failed to destroy service " + serviceKey, t);
  237.         }
  238.     }
  240.     @Override
  241.     public synchronized void notify(List<URL> urls) {
  242.         // 根据 URL 的分类或协议，分组成三个集合 。
  243.         List<URL> invokerUrls = new ArrayList<URL>(); // 服务提供者 URL 集合
  244.         List<URL> routerUrls = new ArrayList<URL>();
  245.         List<URL> configuratorUrls = new ArrayList<URL>();
  246.         for (URL url : urls) {
  247.             String protocol = url.getProtocol();
  248.             String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
  249.             if (Constants.ROUTERS_CATEGORY.equals(category) || Constants.ROUTE_PROTOCOL.equals(protocol)) {
  250.                 routerUrls.add(url);
  251.             } else if (Constants.CONFIGURATORS_CATEGORY.equals(category) || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
  252.                 configuratorUrls.add(url);
  253.             } else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
  254.                 invokerUrls.add(url);
  255.             } else {
  256.                 logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
  257.             }
  258.         }
  259.         // 处理配置规则 URL 集合
  260.         // configurators
  261.         if (!configuratorUrls.isEmpty()) {
  262.             this.configurators = toConfigurators(configuratorUrls);
  263.         }
  264.         // 处理路由规则 URL 集合
  265.         // routers
  266.         if (!routerUrls.isEmpty()) {
  267.             List<Router> routers = toRouters(routerUrls);
  268.             if (routers != null) { // null - do nothing
  269.                 setRouters(routers);
  270.             }
  271.         }
  272.         // 合并配置规则，到 `directoryUrl` 中，形成 `overrideDirectoryUrl` 变量。
  273.         List<Configurator> localConfigurators = this.configurators; // local reference
  274.         // merge override parameters
  275.         this.overrideDirectoryUrl = directoryUrl;
  276.         if (localConfigurators != null && !localConfigurators.isEmpty()) {
  277.             for (Configurator configurator : localConfigurators) {
  278.                 this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
  279.             }
  280.         }
  281.         // 处理服务提供者 URL 集合
  282.         // providers
  283.         refreshInvoker(invokerUrls);
  284.     }
  286.     /**
  287.      * 根据invokerURL列表转换为invoker列表。转换规则如下：
  288.      *
  289.      * 1.如果url已经被转换为invoker，则不在重新引用，直接从缓存中获取，注意如果url中任何一个参数变更也会重新引用
  290.      * 2.如果传入的invoker列表不为空，则表示最新的invoker列表
  291.      * 3.如果传入的invokerUrl列表是空，则表示只是下发的override规则或route规则，需要重新交叉对比，决定是否需要重新引用。
  292.      *
  293.      * @param invokerUrls 传入的参数不能为null
  294.      */
  295.     // TODO: 2017/8/31 FIXME The thread pool should be used to refresh the address, otherwise the task may be accumulated.
  296.     private void refreshInvoker(List<URL> invokerUrls) {
  297.         if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
  298.                 && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
  299.             // 设置禁止访问
  300.             this.forbidden = true; // Forbid to access
  301.             // methodInvokerMap 置空
  302.             this.methodInvokerMap = null; // Set the method invoker map to null
  303.             // 销毁所有 Invoker 集合
  304.             destroyAllInvokers(); // Close all invokers
  305.         } else {
  306.             // 设置允许访问
  307.             this.forbidden = false; // Allow to access
  308.             // 引用老的 urlInvokerMap
  309.             Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
  310.             // 传入的 invokerUrls 为空，说明是路由规则或配置规则发生改变，此时 invokerUrls 是空的，直接使用 cachedInvokerUrls 。
  311.             if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
  312.                 invokerUrls.addAll(this.cachedInvokerUrls);
  313.             // 传入的 invokerUrls 非空，更新 cachedInvokerUrls 。
  314.             } else {
  315.                 this.cachedInvokerUrls = new HashSet<URL>();
  316.                 this.cachedInvokerUrls.addAll(invokerUrls); //Cached invoker urls, convenient for comparison //缓存invokerUrls列表，便于交叉对比
  317.             }
  318.             // 忽略，若无 invokerUrls
  319.             if (invokerUrls.isEmpty()) {
  320.                 return;
  321.             }
  322.             // 将传入的 invokerUrls ，转成新的 urlInvokerMap
  323.             Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
  324.             // 转换出新的 methodInvokerMap
  325.             Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // Change method name to map Invoker Map
  326.             // state change
  327.             // If the calculation is wrong, it is not processed. 如果计算错误，则不进行处理.
  328.             if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
  329.                 logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls.toString()));
  330.                 return;
  331.             }
  332.             // 若服务引用多 group ，则按照 method + group 聚合 Invoker 集合
  333.             this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
  334.             this.urlInvokerMap = newUrlInvokerMap;
  335.             // 销毁不再使用的 Invoker 集合
  336.             try {
  337.                 destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
  338.             } catch (Exception e) {
  339.                 logger.warn("destroyUnusedInvokers error. ", e);
  340.             }
  341.         }
  342.     }
  344.     /**
  345.      * 若服务引用多 group ，则按照 method + group 聚合 Invoker 集合
  346.      */
  347.     private Map<String, List<Invoker<T>>> toMergeMethodInvokerMap(Map<String, List<Invoker<T>>> methodMap) {
  348.         Map<String, List<Invoker<T>>> result = new HashMap<String, List<Invoker<T>>>();
  349.         // 循环方法，按照 method + group 聚合 Invoker 集合
  350.         for (Map.Entry<String, List<Invoker<T>>> entry : methodMap.entrySet()) {
  351.             String method = entry.getKey();
  352.             List<Invoker<T>> invokers = entry.getValue();
  353.             // 按照 Group 聚合 Invoker 集合的结果。其中，KEY：group VALUE：Invoker 集合。
  354.             Map<String, List<Invoker<T>>> groupMap = new HashMap<String, List<Invoker<T>>>();
  355.             // 循环 Invoker 集合，按照 group 聚合 Invoker 集合
  356.             for (Invoker<T> invoker : invokers) {
  357.                 String group = invoker.getUrl().getParameter(Constants.GROUP_KEY, "");
  358.                 List<Invoker<T>> groupInvokers = groupMap.get(group);
  359.                 if (groupInvokers == null) {
  360.                     groupInvokers = new ArrayList<Invoker<T>>();
  361.                     groupMap.put(group, groupInvokers);
  362.                 }
  363.                 groupInvokers.add(invoker);
  364.             }
  365.             // 大小为 1，使用第一个
  366.             if (groupMap.size() == 1) {
  367.                 result.put(method, groupMap.values().iterator().next());
  368.             // 大于 1，将每个 Group 的 Invoker 集合，创建成 Cluster Invoker 对象。
  369.             } else if (groupMap.size() > 1) {
  370.                 List<Invoker<T>> groupInvokers = new ArrayList<Invoker<T>>();
  371.                 for (List<Invoker<T>> groupList : groupMap.values()) {
  372.                     groupInvokers.add(cluster.join(new StaticDirectory<T>(groupList)));
  373.                 }
  374.                 result.put(method, groupInvokers);
  375.             // 大小为 0 ，使用原有值
  376.             } else {
  377.                 result.put(method, invokers);
  378.             }
  379.         }
  380.         return result;
  381.     }
  383.     /**
  384.      * @param urls
  385.      * @return null : no routers ,do nothing
  386.      * else :routers list
  387.      */
  388.     private List<Router> toRouters(List<URL> urls) {
  389.         List<Router> routers = new ArrayList<Router>();
  390.         if (urls == null || urls.isEmpty()) {
  391.             return routers;
  392.         }
  393.         for (URL url : urls) {
  394.             // 忽略，若是 "empty://" 。一般情况下，所有路由规则被删除时，有且仅有一条协议为 "empty://" 的路由规则 URL
  395.             if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
  396.                 continue;
  397.             }
  398.             // 获得 "router"
  399.             String routerType = url.getParameter(Constants.ROUTER_KEY);
  400.             if (routerType != null && routerType.length() > 0) {
  401.                 url = url.setProtocol(routerType);
  402.             }
  403.             try {
  404.                 // 创建 Router 对象
  405.                 Router router = routerFactory.getRouter(url);
  406.                 // 添加到返回结果
  407.                 if (!routers.contains(router)) {
  408.                     routers.add(router);
  409.                 }
  410.             } catch (Throwable t) {
  411.                 logger.error("convert router url to router error, url: " + url, t);
  412.             }
  413.         }
  414.         return routers;
  415.     }
  417.     /**
  418.      * 将服务提供者 URL 集合，转成 Invoker 集合。若该服务提供者 URL 已经转换，则直接复用，不重新引用。
  419.      *
  420.      * @param urls URL 集合
  421.      * @return invokers
  422.      */
  423.     private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
  424.         // 新的 `newUrlInvokerMap`
  425.         Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
  426.         // 若为空，直接返回
  427.         if (urls == null || urls.isEmpty()) {
  428.             return newUrlInvokerMap;
  429.         }
  430.         // 已初始化的服务器提供 URL 集合
  431.         Set<String> keys = new HashSet<String>();
  432.         // 获得引用服务的协议
  433.         String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
  434.         // 循环服务提供者 URL 集合，转成 Invoker 集合
  435.         for (URL providerUrl : urls) {
  436.             // If protocol is configured at the reference side, only the matching protocol is selected
  437.             // 如果 reference 端配置了 protocol ，则只选择匹配的 protocol
  438.             if (queryProtocols != null && queryProtocols.length() > 0) {
  439.                 boolean accept = false;
  440.                 String[] acceptProtocols = queryProtocols.split(","); // 可配置多个协议
  441.                 for (String acceptProtocol : acceptProtocols) {
  442.                     if (providerUrl.getProtocol().equals(acceptProtocol)) {
  443.                         accept = true;
  444.                         break;
  445.                     }
  446.                 }
  447.                 if (!accept) {
  448.                     continue;
  449.                 }
  450.             }
  451.             // 忽略，若为 `empty://` 协议
  452.             if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
  453.                 continue;
  454.             }
  455.             // 忽略，若应用程序不支持该协议
  456.             if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
  457.                 logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() + " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost()
  458.                         + ", supported protocol: " + ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
  459.                 continue;
  460.             }
  461.             // 合并 URL 参数
  462.             URL url = mergeUrl(providerUrl);
  463.             // 忽略，若已经初始化
  464.             String key = url.toFullString(); // The parameter urls are sorted
  465.             if (keys.contains(key)) { // Repeated url
  466.                 continue;
  467.             }
  468.             // 添加到 `keys` 中
  469.             keys.add(key);
  470.             // Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
  471.             // 如果服务端 URL 发生变化，则重新 refer 引用
  472.             Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
  473.             Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
  474.             if (invoker == null) { // Not in the cache, refer again 未在缓存中，重新引用
  475.                 try {
  476.                     // 判断是否开启
  477.                     boolean enabled;
  478.                     if (url.hasParameter(Constants.DISABLED_KEY)) {
  479.                         enabled = !url.getParameter(Constants.DISABLED_KEY, false);
  480.                     } else {
  481.                         enabled = url.getParameter(Constants.ENABLED_KEY, true);
  482.                     }
  483.                     // 若开启，创建 Invoker 对象
  484.                     if (enabled) {
  485.                         // 注意，引用服务
  486.                         invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
  487.                     }
  488.                 } catch (Throwable t) {
  489.                     logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
  490.                 }
  491.                 // 添加到 newUrlInvokerMap 中
  492.                 if (invoker != null) { // Put new invoker in cache
  493.                     newUrlInvokerMap.put(key, invoker);
  494.                 }
  495.             } else { // 在缓存中，直接使用缓存的 Invoker 对象，添加到 newUrlInvokerMap 中
  496.                 newUrlInvokerMap.put(key, invoker);
  497.             }
  498.         }
  499.         // 清空 keys
  500.         keys.clear();
  501.         return newUrlInvokerMap;
  502.     }
  504.     /**
  505.      * Merge url parameters. the order is: override > -D >Consumer > Provider
  506.      *
  507.      * 合并 URL 参数，优先级为配置规则 > 服务消费者配置 > 服务提供者配置
  508.      *
  509.      * @param providerUrl 服务提供者 URL
  510.      * @return 合并后的 URL
  511.      */
  512.     private URL mergeUrl(URL providerUrl) {
  513.         // 合并消费端参数
  514.         providerUrl = ClusterUtils.mergeUrl(providerUrl, queryMap); // Merge the consumer side parameters
  516.         // 合并配置规则
  517.         List<Configurator> localConfigurators = this.configurators; // local reference
  518.         if (localConfigurators != null && !localConfigurators.isEmpty()) {
  519.             for (Configurator configurator : localConfigurators) {
  520.                 providerUrl = configurator.configure(providerUrl);
  521.             }
  522.         }
  524.         // 不检查连接是否成功，总是创建 Invoker ！因为，启动检查，只有启动阶段需要。此时在检查，已经没必要了。
  525.         providerUrl = providerUrl.addParameter(Constants.CHECK_KEY, String.valueOf(false)); // Do not check whether the connection is successful or not, always create Invoker!
  527.         // The combination of directoryUrl and override is at the end of notify, which can't be handled here
  528.         // 仅合并提供者参数，因为 directoryUrl 与 override 合并是在 notify 的最后，这里不能够处理
  529.         this.overrideDirectoryUrl = this.overrideDirectoryUrl.addParametersIfAbsent(providerUrl.getParameters()); // Merge the provider side parameters
  531.         // 【忽略】因为是对 1.0 版本的兼容
  532.         if ((providerUrl.getPath() == null || providerUrl.getPath().length() == 0)
  533.                 && "dubbo".equals(providerUrl.getProtocol())) { // Compatible version 1.0
  534.             //fix by tony.chenl DUBBO-44
  535.             String path = directoryUrl.getParameter(Constants.INTERFACE_KEY);
  536.             if (path != null) {
  537.                 int i = path.indexOf('/');
  538.                 if (i >= 0) {
  539.                     path = path.substring(i + 1);
  540.                 }
  541.                 i = path.lastIndexOf(':');
  542.                 if (i >= 0) {
  543.                     path = path.substring(0, i);
  544.                 }
  545.                 providerUrl = providerUrl.setPath(path);
  546.             }
  547.         }
  549.         // 返回服务提供者 URL
  550.         return providerUrl;
  551.     }
  553.     private List<Invoker<T>> route(List<Invoker<T>> invokers, String method) {
  554.         // 创建 Invocation 对象
  555.         Invocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
  556.         // 获得 Router 数组
  557.         List<Router> routers = getRouters();
  558.         // 根据路由规则，筛选 Invoker 集合
  559.         if (routers != null) {
  560.             for (Router router : routers) {
  561.                 if (router.getUrl() != null) {
  562.                     invokers = router.route(invokers, getConsumerUrl(), invocation);
  563.                 }
  564.             }
  565.         }
  566.         return invokers;
  567.     }
  569.     /**
  570.      * 将invokers列表转成与方法的映射关系
  571.      *
  572.      * @param invokersMap Invoker列表
  573.      * @return Invoker与方法的映射关系
  574.      */
  575.     private Map<String, List<Invoker<T>>> toMethodInvokers(Map<String, Invoker<T>> invokersMap) {
  576.         // 创建新的 `methodInvokerMap`
  577.         Map<String, List<Invoker<T>>> newMethodInvokerMap = new HashMap<String, List<Invoker<T>>>();
  578.         // 创建 Invoker 集合
  579.         List<Invoker<T>> invokersList = new ArrayList<Invoker<T>>();
  580.         // According to the methods classification declared by the provider URL, the methods is compatible with the registry to execute the filtered methods
  581.         // 按服务提供者 URL 所声明的 methods 分类，兼容注册中心执行路由过滤掉的 methods
  582.         if (invokersMap != null && invokersMap.size() > 0) {
  583.             // 循环每个服务提供者 Invoker
  584.             for (Invoker<T> invoker : invokersMap.values()) {
  585.                 String parameter = invoker.getUrl().getParameter(Constants.METHODS_KEY); // methods
  586.                 if (parameter != null && parameter.length() > 0) {
  587.                     String[] methods = Constants.COMMA_SPLIT_PATTERN.split(parameter);
  588.                     if (methods != null && methods.length > 0) {
  589.                         // 循环每个方法，按照方法名为维度，聚合到 `methodInvokerMap` 中
  590.                         for (String method : methods) {
  591.                             if (method != null && method.length() > 0 && !Constants.ANY_VALUE.equals(method)) { // 当服务提供者的方法为 "*" ，代表泛化调用
  592.                                 List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
  593.                                 if (methodInvokers == null) {
  594.                                     methodInvokers = new ArrayList<Invoker<T>>();
  595.                                     newMethodInvokerMap.put(method, methodInvokers);
  596.                                 }
  597.                                 methodInvokers.add(invoker);
  598.                             }
  599.                         }
  600.                     }
  601.                 }
  602.                 // 添加到 `invokersList` 中
  603.                 invokersList.add(invoker);
  604.             }
  605.         }
  606.         // 路由全 `invokersList` ，匹配合适的 Invoker 集合
  607.         List<Invoker<T>> newInvokersList = route(invokersList, null);
  608.         // 添加 `newInvokersList` 到 `newMethodInvokerMap` 中，表示该服务提供者的全量 Invoker 集合
  609.         newMethodInvokerMap.put(Constants.ANY_VALUE, newInvokersList);
  610.         // 循环，基于每个方法路由，匹配合适的 Invoker 集合
  611.         if (serviceMethods != null && serviceMethods.length > 0) {
  612.             for (String method : serviceMethods) {
  613.                 List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
  614.                 if (methodInvokers == null || methodInvokers.isEmpty()) {
  615.                     methodInvokers = newInvokersList;
  616.                 }
  617.                 newMethodInvokerMap.put(method, route(methodInvokers, method));
  618.             }
  619.         }
  620.         // 循环排序每个方法的 Invoker 集合，并设置为不可变
  621.         // sort and unmodifiable
  622.         for (String method : new HashSet<String>(newMethodInvokerMap.keySet())) {
  623.             List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
  624.             Collections.sort(methodInvokers, InvokerComparator.getComparator());
  625.             newMethodInvokerMap.put(method, Collections.unmodifiableList(methodInvokers));
  626.         }
  627.         return Collections.unmodifiableMap(newMethodInvokerMap);
  628.     }
  630.     /**
  631.      * Close all invokers
  632.      */
  633.     private void destroyAllInvokers() {
  634.         Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference 本地引用，避免并发问题
  635.         if (localUrlInvokerMap != null) {
  636.             // 循环 urlInvokerMap ，销毁所有服务提供者 Invoker
  637.             for (Invoker<T> invoker : new ArrayList<Invoker<T>>(localUrlInvokerMap.values())) {
  638.                 try {
  639.                     invoker.destroy();
  640.                 } catch (Throwable t) {
  641.                     logger.warn("Failed to destroy service " + serviceKey + " to provider " + invoker.getUrl(), t);
  642.                 }
  643.             }
  644.             // urlInvokerMap 清空
  645.             localUrlInvokerMap.clear();
  646.         }
  647.         // methodInvokerMap 置空
  648.         methodInvokerMap = null;
  649.     }
  651.     /**
  652.      * Check whether the invoker in the cache needs to be destroyed
  653.      * If set attribute of url: refer.autodestroy=false, the invokers will only increase without decreasing,there may be a refer leak
  654.      *
  655.      * @param oldUrlInvokerMap
  656.      * @param newUrlInvokerMap
  657.      */
  658.     private void destroyUnusedInvokers(Map<String, Invoker<T>> oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {
  659.         // 防御性编程，目前不存在这个情况
  660.         if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
  661.             // 销毁所有服务提供者 Invoker
  662.             destroyAllInvokers();
  663.             return;
  664.         }
  665.         // check deleted invoker
  666.         // 对比新老集合，计算需要销毁的 Invoker 集合
  667.         List<String> deleted = null;
  668.         if (oldUrlInvokerMap != null) {
  669.             Collection<Invoker<T>> newInvokers = newUrlInvokerMap.values();
  670.             for (Map.Entry<String, Invoker<T>> entry : oldUrlInvokerMap.entrySet()) {
  671.                 // 若不存在，添加到 `deleted` 中
  672.                 if (!newInvokers.contains(entry.getValue())) {
  673.                     if (deleted == null) {
  674.                         deleted = new ArrayList<String>();
  675.                     }
  676.                     deleted.add(entry.getKey());
  677.                 }
  678.             }
  679.         }
  681.         // 若有需要销毁的 Invoker ，则进行销毁
  682.         if (deleted != null) {
  683.             for (String url : deleted) {
  684.                 if (url != null) {
  685.                     // 移除出 `urlInvokerMap`
  686.                     Invoker<T> invoker = oldUrlInvokerMap.remove(url);
  687.                     if (invoker != null) {
  688.                         try {
  689.                             // 销毁 Invoker
  690.                             invoker.destroy();
  691.                             if (logger.isDebugEnabled()) {
  692.                                 logger.debug("destroy invoker[" + invoker.getUrl() + "] success. ");
  693.                             }
  694.                         } catch (Exception e) {
  695.                             logger.warn("destroy invoker[" + invoker.getUrl() + "] failed. " + e.getMessage(), e);
  696.                         }
  697.                     }
  698.                 }
  699.             }
  700.         }
  701.     }
  703.     @Override
  704.     public List<Invoker<T>> doList(Invocation invocation) {
  705.         if (forbidden) {
  706.             // 1. No service provider 2. Service providers are disabled
  707.             throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
  708.                 "No provider available from registry " + getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " +  NetUtils.getLocalHost()
  709.                         + " use dubbo version " + Version.getVersion() + ", please check status of providers(disabled, not registered or in blacklist).");
  710.         }
  711.         List<Invoker<T>> invokers = null;
  712.         Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
  713.         // 获得 Invoker 集合
  714.         if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
  715.             // 获得方法名、方法参数
  716.             String methodName = RpcUtils.getMethodName(invocation);
  717.             Object[] args = RpcUtils.getArguments(invocation);
  718.             // 【第一】可根据第一个参数枚举路由
  719.             if (args != null && args.length > 0 && args[0] != null
  720.                     && (args[0] instanceof String || args[0].getClass().isEnum())) {
  721. //                invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
  722.                 invokers = localMethodInvokerMap.get(methodName + args[0]); // The routing can be enumerated according to the first parameter
  723.             }
  724.             // 【第二】根据方法名获得 Invoker 集合
  725.             if (invokers == null) {
  726.                 invokers = localMethodInvokerMap.get(methodName);
  727.             }
  728.             // 【第三】使用全量 Invoker 集合。例如，`#$echo(name)` ，回声方法
  729.             if (invokers == null) {
  730.                 invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
  731.             }
  732.             // 【第四】使用 `methodInvokerMap` 第一个 Invoker 集合。防御性编程。
  733.             if (invokers == null) {
  734.                 Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
  735.                 if (iterator.hasNext()) {
  736.                     invokers = iterator.next();
  737.                 }
  738.             }
  739.         }
  740.         return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
  741.     }
  743.     @Override
  744.     public Class<T> getInterface() {
  745.         return serviceType;
  746.     }
  748.     @Override
  749.     public URL getUrl() {
  750.         return this.overrideDirectoryUrl;
  751.     }
  753.     @Override
  754.     public boolean isAvailable() {
  755.         // 若已销毁，返回不可用
  756.         if (isDestroyed()) {
  757.             return false;
  758.         }
  759.         // 任意一个 Invoker 可用，则返回可用
  760.         Map<String, Invoker<T>> localUrlInvokerMap = urlInvokerMap;
  761.         if (localUrlInvokerMap != null && localUrlInvokerMap.size() > 0) {
  762.             for (Invoker<T> invoker : new ArrayList<Invoker<T>>(localUrlInvokerMap.values())) {
  763.                 if (invoker.isAvailable()) {
  764.                     return true;
  765.                 }
  766.             }
  767.         }
  768.         return false;
  769.     }
  771.     /**
  772.      * Haomin: added for test purpose
  773.      */
  774.     public Map<String, Invoker<T>> getUrlInvokerMap() {
  775.         return urlInvokerMap;
  776.     }
  778.     /**
  779.      * Haomin: added for test purpose
  780.      */
  781.     public Map<String, List<Invoker<T>>> getMethodInvokerMap() {
  782.         return methodInvokerMap;
  783.     }
  785.     /**
  786.      * Invoker 排序器，根据 URL 升序
  787.      */
  788.     private static class InvokerComparator implements Comparator<Invoker<?>> {
  790.         /**
  791.          * 单例
  792.          */
  793.         private static final InvokerComparator comparator = new InvokerComparator();
  795.         private InvokerComparator() {
  796.         }
  798.         public static InvokerComparator getComparator() {
  799.             return comparator;
  800.         }
  802.         @Override
  803.         public int compare(Invoker<?> o1, Invoker<?> o2) {
  804.             return o1.getUrl().toString().compareTo(o2.getUrl().toString());
  805.         }
  807.     }
  809.     /**
  810.      *
  811.      * Invoker 代理类，主要用于存储注册中心下发的 url 地址，用于重新重新 refer 时能够根据 providerURL queryMap overrideMap 重新组装
  812.      *
  813.      * @param <T>
  814.      */
  815.     private static class InvokerDelegate<T> extends InvokerWrapper<T> {
  817.         /**
  818.          * 服务提供者 URL
  819.          *
  820.          * 未经过配置合并
  821.          */
  822.         private URL providerUrl;
  824.         public InvokerDelegate(Invoker<T> invoker, URL url, URL providerUrl) {
  825.             super(invoker, url);
  826.             this.providerUrl = providerUrl;
  827.         }
  829.         public URL getProviderUrl() {
  830.             return providerUrl;
  831.         }
  832.     }
  833. }