Dubbo3 AbstractRegistry源码分析

3.8.4 核心流程

✅ 1. 初始化流程

• 解析配置URL，设置基本参数
• 创建本地缓存文件路径
• 加载已有缓存数据
• 启动定时保存任务

✅ 2. 注册流程

• 验证URL有效性
• 添加到已注册集合
• 记录日志信息

✅ 3. 订阅流程

• 验证参数有效性
• 建立订阅关系
• 触发首次通知

✅ 4. 通知流程

• 匹配订阅者URL
• 分类整理通知数据
• 调用监听器通知方法
• 更新本地缓存

✅ 5. 销毁流程

• 注销所有已注册服务
• 取消所有订阅关系
• 清理本地资源

3.8.5 设计特点

• 抽象基类: 提供通用实现，具体注册中心可继承并实现特定逻辑
• SPI支持: 符合Dubbo的SPI扩展机制
• 线程安全: 使用并发集合和同步机制保证线程安全
• 容错设计: 提供完善的异常处理和恢复机制
• 性能优化: 使用延迟保存和批量处理提高性能

3.8.6 本地缓存

本地缓存将被保存在Properties文件当中

// Local disk cache, where the special key value.registries records the list of registry centers, and the others are
// the list of notified service providers
private final Properties properties = new Properties();

// Local disk cache file
private File file;

• 本地的 Properties 文件缓存，properties 是加载到内存的 Properties 对象

• file 是磁盘上对应的文件，两者的数据是同步的。

✅ 1. 核心组件

• properties: 内存中的属性对象，用于存储缓存数据
• file: 本地磁盘缓存文件，持久化存储注册数据
• lastCacheChanged: 缓存版本号，用于控制并发写入
• registryCacheExecutor: 定时任务执行器，用于异步保存缓存

✅ 2. 缓存初始化流程

• 根据配置确定缓存文件路径，默认为用户主目录下的 dubbo-registry 文件夹
• 创建缓存文件及其父目录
• 启动时加载已有缓存文件内容到 properties 对象中
• 通过 loadProperties() 方法读取本地缓存数据

✅ 3. 缓存的数据结构

• 缓存以键值对形式存储：
  • 键: 服务键（service key），由接口名、版本号和分组构成
  • 值: URL字符串列表，以空格分隔的URL全字符串表示

✅ 4. 缓存写入机制

• 异步保存

public void doSaveProperties(long version) {
	// ....
    // 通过定时任务异步保存
    if (version < lastCacheChanged.get()) {
        savePropertiesRetryTimes.set(0);
        return;
    } else {
        registryCacheExecutor.schedule(
            () -> doSaveProperties(lastCacheChanged.incrementAndGet()),
            DEFAULT_INTERVAL_SAVE_PROPERTIES,
            TimeUnit.MILLISECONDS);
    }
}

• 同步保存

• 当配置 syncSaveFile=true 时，直接在当前线程同步写入文件
• 文件锁机制: 为保证多进程环境下的数据一致性，采用文件锁保护：
  • 创建与缓存文件同名的 .lock 文件
  • 使用 FileChannel.tryLock() 获取文件锁
  • 成功获取锁后才进行写入操作
  • 写入完成后释放文件锁并删除锁文件

✅ 5. 缓存更新触发点

1. 通知时更新: 每次 notify() 调用后自动更新相关URL的缓存
2. 版本控制: 通过 lastCacheChanged 版本号避免过期数据覆盖新数据
3. 重试机制: 失败时最多重试3次，防止因锁竞争导致的数据丢失

✅ 6. 缓存读取策略

• 优先从内存 notified 映射中获取数据
• 如果内存中没有，则从 properties 属性对象中查找
• 启动时通过 loadProperties() 预加载历史数据

这种双重缓存机制（内存+磁盘）确保了即使在网络故障或注册中心宕机的情况下，消费者和提供者仍然可以基于本地缓存进行服务发现和调用。

✅ 7. 图解

• 缓存的写入流程

• 缓存的读取流程

✅ 8. notify方法

作为一个 RPC 框架，Dubbo 在微服务架构中解决了各个服务间协作的难题；

作为 Provider 和 Consumer 的底层依赖，它会与服务一起打包部署。dubbo-registry 也仅仅是其中一个依赖包，负责完成与 ZooKeeper、etcd、Consul 等服务发现组件的交互。

当 Provider端暴露的 URL 发生变化时，Nacos等服务发现组件会通知 Consumer 端的 Registry 组件，Registry 组件会调用 notify() 方法，被通知的 Consumer 能匹配到所有 Provider 的 URL 列表并写入 properties 集合中。

notify方法核心实现

/**
     * Notify changes from the provider side.
     *
     * @param url      消费者端的订阅URL，用于匹配和过滤通知的URL
     * @param listener 通知监听器，实际接收通知的对象
     * @param urls     提供者端最新的URL列表，需要通知给监听器
     */
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
 	// 首先验证必要参数不能为null，确保方法调用的有效性。
    // 空URL处理: 如果传入的URL列表为空且不是通配符订阅，则记录警告日志并返回，避免无效通知。
    if ((CollectionUtils.isEmpty(urls)) && !ANY_VALUE.equals(url.getServiceInterface())) {
        // 1-4 Empty address.
        logger.warn(REGISTRY_EMPTY_ADDRESS, "", "", "Ignore empty notify urls for subscribe url " + url);
        return;
    }
    
	
    // 遍历所有URL，使用 UrlUtils.isMatch() 方法筛选出匹配当前订阅URL的URL，
    // 并按照category（默认为providers）进行分类存储。
    // keep every provider's category.
    Map<String, List<URL>> result = new HashMap<>();
    for (URL u : urls) {
        if (UrlUtils.isMatch(url, u)) {
            String category = u.getCategory(DEFAULT_CATEGORY);
            List<URL> categoryList = result.computeIfAbsent(category, k -> new ArrayList<>());
            categoryList.add(u);
        }
    }
    if (result.size() == 0) {
        return;
    }
    
    // 通知监听器: 逐个category通知监听器，并更新本地缓存。
    Map<String, List<URL>> categoryNotified =
        ConcurrentHashMapUtils.computeIfAbsent(notified, url, u -> new ConcurrentHashMap<>());
    for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
        String category = entry.getKey();
        List<URL> categoryList = entry.getValue();
        categoryNotified.put(category, categoryList);
        listener.notify(categoryList);
        if (localCacheEnabled) {
            saveProperties(url);
        }
    }
}

🔝 关键特性一: URL匹配机制

if (UrlUtils.isMatch(url, u)) {
    String category = u.getCategory(DEFAULT_CATEGORY);
    List<URL> categoryList = result.computeIfAbsent(category, k -> new ArrayList<>());
    categoryList.add(u);
}

使用 UrlUtils.isMatch() 方法精确匹配订阅URL和提供者URL，确保只有符合订阅条件的URL才会被通知。

 public static boolean isMatch(URL consumerUrl, URL providerUrl) {
        String consumerInterface = consumerUrl.getServiceInterface();
        String providerInterface = providerUrl.getServiceInterface();
     // ...
 }

匹配规则:

• ✅ 接口匹配规则

  • 消费者URL和提供者URL的接口必须匹配
  • 支持通配符匹配：如果任一URL的接口为*（ANY_VALUE），则接口匹配成功
  • 精确匹配：两个URL的接口名称必须完全相同

• ✅ 分类匹配规则

  • 检查提供者URL的分类（category）是否与消费者URL的分类匹配
  • 默认分类为providers
  • 支持多种分类匹配模式：
    • 如果消费者分类包含*，则匹配所有分类
    • 如果消费者分类以-开头，表示排除该分类
    • 精确匹配消费者指定的分类

• ✅ 使能状态匹配规则

  • 提供者URL必须是启用状态（enabled=true）
  • 如果提供者URL的enabled参数为false，则匹配失败
  • 消费者URL可以显式接受禁用的服务（通过设置enabled参数

• ✅ 分组（Group）匹配规则

  • 消费者分组为*时匹配所有提供者分组
  • 消费者分组与提供者分组精确匹配
  • 消费者分组包含提供者分组（通过StringUtils.isContains方法）

• ✅ 版本（Version）匹配规则

  • 消费者版本为*时匹配所有提供者版本
  • 消费者版本与提供者版本必须精确匹配

• ✅ 分类器（Classifier）匹配规则

  • 消费者分类器为null或*时匹配所有提供者分类器
  • 消费者分类器与提供者分类器必须精确匹配

所有上述条件必须同时满足，即接口、分类、使能状态、分组、版本和分类器都匹配时，方法才返回true。任何一个条件不满足都会导致匹配失败。

⬆️ 2. 关键特性2: 分类通知

支持按category分类通知, 包括

• providers（服务提供者）
• consumers（服务消费者）
• routers（路由规则）
• configurators（配置规则）

➡️ 3. 设计考虑

• 线程安全: 使用ConcurrentHashMap等线程安全的数据结构
• 性能优化: 通过分类存储减少重复匹配计算
• 容错机制: 即使通知失败也不会影响其他监听器
• 缓存一致性: 保证内存缓存和磁盘缓存的一致性
• 这个方法是Dubbo服务发现机制的核心，负责将注册中心的服务变化及时准确地通知给所有订阅者。

这个方法是Dubbo服务发现机制的核心，负责将注册中心的服务变化及时准确地通知给所有订阅者。

🎉🎉🎉 enjoy it!