David Liu's Blog 首页归档标签分类
Soul 网关源码学习(7) - 从依赖的角度理解 Soul 网关 (3) - Zookeeper 数据同步机制
|字数总计: 1.9k|阅读时长: 7 分钟

soul admin 和 soul bootstrap 同步数据的时机

从昨天的文章中可以看出 soul admin 和 soul bootstrap 之间数据同步的触发的类型主要有两类行为:

  1. soul admin 中对 soul bootstrap 运行时数据 (app_auth, plugin, selector, rule 等) 进行了更改，由 soul admin 侧的事件机制触发数据的同步
  2. soul bootstrap 主动触发数据同步，如：WebSocket 客户端和服务端完成了握手行为 (WebSocket 协议应用层的属于) 之后由 soul bootstrap 端主动触发数据的同步

从源码分析 soul 网关如何依赖 zookeeper 完成数据同步

同步机制

soul admin 利用 zookeeper 同步数据的原理是使用了 zookeeper 的 watcher 机制, zookeeper 的 watcher 流程如下:

  • 客户端向服务端的某个节点路径上注册一个 watcher，客户端同时会在本地 watcher manager 中存储特定的watcher
  • 当发生节点数据或者节点子节点变化时，服务端会通知客户端节点变化信息，客户端收到通知后，会调用 watcher 回调函数

在 soul 网关中, soul admin 中集成了 zookeeper 客户端, 在 soul admin 中的数据类型变更数据处理流程中和 zookeeper server 交互; soul bootstrap 模块可配置启用通过 zookeeper 同步数据, 通过 zookeeper 同步数据的功能被集成在了模块
soul-spring-boot-starter-sync-data-zookeeper 中, 这个模块中有注册 watcher 到 zookeeper 节点路径上的逻辑; 通过上述的流程完成数据的同步。

soul admin 侧的流程解析

soul admin 工程中，在 Spring 事件总线机制中完成 soul 网关数据同步事件监听的类是: org.dromara.soul.admin.listener.DataChangedListener, zookeeper
完成数据同步的实现类为: org.dromara.soul.admin.listener.zookeeper.ZookeeperDataChangedListener:

public class ZookeeperDataChangedListener implements DataChangedListener {

    private final ZkClient zkClient;

    public ZookeeperDataChangedListener(final ZkClient zkClient) {
        this.zkClient = zkClient;
    }

    @Override
    public void onAppAuthChanged(final List<AppAuthData> changed, final DataEventTypeEnum eventType) {
        for (AppAuthData data : changed) {
            final String appAuthPath = ZkPathConstants.buildAppAuthPath(data.getAppKey()); // 构造 app auth data 的 zookeeper 数据存储节点的路径信息
            // delete
            if (eventType == DataEventTypeEnum.DELETE) {
                deleteZkPath(appAuthPath); // 如果事件的操作类型是删除，则删除 zookeeper 中的节点
                continue;
            }
            // create or update
            upsertZkNode(appAuthPath, data); // 创建或者更新 zookeeper 中节点的信息 (将数据信息写入到 zookeeper 节点中)
        }
    }

    @SneakyThrows
    @Override
    public void onMetaDataChanged(final List<MetaData> changed, final DataEventTypeEnum eventType) {
        for (MetaData data : changed) {
            // 构造 meta data 的 zookeeper 数据存储节点的路径信息
            final String metaDataPath = ZkPathConstants.buildMetaDataPath(URLEncoder.encode(data.getPath(), "UTF-8"));
            // delete
            if (eventType == DataEventTypeEnum.DELETE) {
                deleteZkPath(metaDataPath); // 如果事件的操作类型是删除，则删除 zookeeper 中的节点
                continue;
            }
            // create or update
            upsertZkNode(metaDataPath, data); // 创建或者更新 zookeeper 中节点的信息 (将数据信息写入到 zookeeper 节点中)
        }
    }

    @Override
    public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
        for (PluginData data : changed) {
            // 构造 plugin data 的 zookeeper 数据存储节点的路径信息
            final String pluginPath = ZkPathConstants.buildPluginPath(data.getName());
            // delete
            if (eventType == DataEventTypeEnum.DELETE) {
                // 如果事件的操作类型是删除，则删除 zookeeper 中的节点
                // selector, rule 数据和 plugin 数据存在关联:
                // selector 节点 zookeeper 路径规则为: /soul/selector/{pluginName}/{selectorId} 
                // rule 节点 zookeeper 路径规则为： zoo/soul/rule/{pluginName}/{selectorId}-{ruleId}
                // 所以在操作 zookeeper 中插件路径之后会操作按照规则关联的 selector 路径和 rule 路径
                deleteZkPathRecursive(pluginPath);
                final String selectorParentPath = ZkPathConstants.buildSelectorParentPath(data.getName());
                deleteZkPathRecursive(selectorParentPath);
                final String ruleParentPath = ZkPathConstants.buildRuleParentPath(data.getName());
                deleteZkPathRecursive(ruleParentPath);
                continue;
            }
            //create or update
            upsertZkNode(pluginPath, data); // 创建或者更新 zookeeper 中节点的信息 (将数据信息写入到 zookeeper 节点中)
        }
    }

    @Override
    public void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {
        if (eventType == DataEventTypeEnum.REFRESH) {
            // 如果 DataEventType 类型是 REFRESH
            // 获取 selector parent 路径, selector 的路径规则是: /soul/selector/{pluginName}/ 
            final String selectorParentPath = ZkPathConstants.buildSelectorParentPath(changed.get(0).getPluginName());
            // 删除 selector parent 路径
            deleteZkPathRecursive(selectorParentPath);
        }
        for (SelectorData data : changed) {
            // 构建插件选择器数据存储的数据节点路径
            final String selectorRealPath = ZkPathConstants.buildSelectorRealPath(data.getPluginName(), data.getId());
            if (eventType == DataEventTypeEnum.DELETE) {
                // 如果 DataEventType 类型是 REFRESH, 删除插件选择器数据路径
                deleteZkPath(selectorRealPath);
                continue;
            }
            // 构建插件选择器所属插件的路径
            final String selectorParentPath = ZkPathConstants.buildSelectorParentPath(data.getPluginName());
            createZkNode(selectorParentPath);
            //create or update
            upsertZkNode(selectorRealPath, data); // 创建或者更新 zookeeper 中节点的信息 (将数据信息写入到 zookeeper 节点中)
        }
    }

    @Override
    // 处理逻辑和 selector 数据处理方式类似
    public void onRuleChanged(final List<RuleData> changed, final DataEventTypeEnum eventType) {
        if (eventType == DataEventTypeEnum.REFRESH) {
            final String selectorParentPath = ZkPathConstants.buildRuleParentPath(changed.get(0).getPluginName());
            deleteZkPathRecursive(selectorParentPath);
        }
        for (RuleData data : changed) {
            final String ruleRealPath = ZkPathConstants.buildRulePath(data.getPluginName(), data.getSelectorId(), data.getId());
            if (eventType == DataEventTypeEnum.DELETE) {
                deleteZkPath(ruleRealPath);
                continue;
            }
            final String ruleParentPath = ZkPathConstants.buildRuleParentPath(data.getPluginName());
            createZkNode(ruleParentPath);
            //create or update
            upsertZkNode(ruleRealPath, data);
        }
    }

    private void createZkNode(final String path) {
        if (!zkClient.exists(path)) {
            // 如果路径不存在，创建持久存储节点
            zkClient.createPersistent(path, true);
        }
    }

    /**
     * create or update zookeeper node.
     * @param path node path
     * @param data node data 
     */
    private void upsertZkNode(final String path, final Object data) {
        if (!zkClient.exists(path)) {
            // 如果路径不存在，创建持久存储节点
            zkClient.createPersistent(path, true);
        }
        // 将数据写入 zookeeper 路径中
        zkClient.writeData(path, data);
    }

    private void deleteZkPath(final String path) {
        if (zkClient.exists(path)) {
            // 如果路径存在，删除路径节点
            zkClient.delete(path);
        }
    }

    private void deleteZkPathRecursive(final String path) {
        if (zkClient.exists(path)) {
            // 如果路径存在，递归删除路径节点 (将指定路径下的所有节点删除)
            zkClient.deleteRecursive(path);
        }
    }
}

在 zookeeper 中 plugin 节点, selector 节点, rule 节点的路径有一定的关联，关联关系如下:

路径类型 路径规则
AppAuthData /soul/auth/{appKey}
MetaData /soul/meta/{path}
PluginData /soul/plugin/{pluginName}
SelectorData /soul/selector/{pluginName}/{selectorId}
RuleData /soul/rule/{pluginName}/{selectorId}-{ruleId}

zookeeper 路径的构造方法在: org.dromara.soul.common.constant.ZkPathConstants 中。

所以在 ZookeeperDataChangedListener 类的 onPluginChanged,
onSelectorChanged, onRuleChanged 方法会存在对 zookeeper 路径的复合操作。

soul admin 利用 zookeeper 同步数据到 soul bootstrap 侧依赖的是 soul admin 中监听数据变化事件的处理中的实现类 ZookeeperDataChangedListener
对 zookeeper 中持久节点的数据节点操作完成不同类型数据的操作来完成不用的 DataEventType 的操作。

soul bootstrap 侧的流程剖析

soul bootstrap 集成模块 soul-spring-boot-starter-sync-data-zookeeper 并且在 Spring 容器启动配置中加入相应的配置来开启 zookeeper 配置。

配置 zookeeper 数据同步的关键 Configuration


@Configuration
@ConditionalOnClass(ZookeeperSyncDataService.class)
@ConditionalOnProperty(prefix = "soul.sync.zookeeper", name = "url")
@EnableConfigurationProperties(ZookeeperConfig.class)
@Slf4j
public class ZookeeperSyncDataConfiguration {

    @Bean
    public SyncDataService syncDataService(final ObjectProvider<ZkClient> zkClient, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
                                           final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
        log.info("you use zookeeper sync soul data.......");
        return new ZookeeperSyncDataService(zkClient.getIfAvailable(), pluginSubscriber.getIfAvailable(),
                metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));
    }

    @Bean
    public ZkClient zkClient(final ZookeeperConfig zookeeperConfig) {
        return new ZkClient(zookeeperConfig.getUrl(), zookeeperConfig.getSessionTimeout(), zookeeperConfig.getConnectionTimeout());
    }

}
  • 配置会被 Spring 容器执行的条件是: classpath 中存在 ZookeeperSyncDataService 类并且 PropertySource 中存在配置 soul.sync.zookeeper.url
  • ZookeeperSyncDataConfiguration 配置类中注册的 SyncDataService 为通过 zookeeper 实现 watcher 机制的关键类, ZookeeperSyncDataConfiguration 的构造函数中完执行了:
    • 从 zookeeper 读取插件信息, 选择器信息, 规则信息。
    • 注册 watcher 到插件节点路径, 选择器节点路径, 规则节点路径; 当发生节点数据或者节点子节点变化时，服务端会通知客户端节点变化信息，完成数据的同步。

总结

soul 利用了 zookeeper 路径节点的 watcher 机制来实现 soul admin 到 soul bootstrap 的数据同步, 结合昨天分析的通过 WebSocket 同步数据, 可以看到一些设计模式的灵活运用, 比如: 观察者模式, 策略模式等, 值得细细学习。

文章作者: David Liu
文章链接: https://davidliu.now.sh/2021/01/21/soul_dependency_discoveryII/
版权声明: 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 David Liu's Blog！
soul网关