Soul源码阅读-基于Zookeeper数据同步

Soul Admin

Soul Admin在启动时，将配置数据信息注册到Zookeeper上。当有配置信息变更时，会将变更的信息同步更新到Zookeeper上。

核心类

ZookeeperDataInit

该类在启动时由Spring容器加载，负责将配置数据全量注册到Zookeeper上

public void run(final String... args) {
    // 设置 plugin 的 zookeeper path
    String pluginPath = ZkPathConstants.PLUGIN_PARENT;
    // 设置 auth 的 zookeeper path
    String authPath = ZkPathConstants.APP_AUTH_PARENT;
    // 设置 metadata 的 zookeeper path
    String metaDataPath = ZkPathConstants.META_DATA;
    // plugin、auth和metadata都不存在时进行注册
    if (!zkClient.exists(pluginPath) && !zkClient.exists(authPath) && !zkClient.exists(metaDataPath)) {
        // 从DB获取全量数据然后进行注册
        // DataEventTypeEnum.REFRESH 代表全量刷新
        syncDataService.syncAll(DataEventTypeEnum.REFRESH);
    }
}

SyncDataServiceImpl

负责从DB获取数据并发布数据变更事件。

public boolean syncAll(final DataEventTypeEnum type) {
    // 从DB获取授权认证配置数据，并发布事件
    appAuthService.syncData();
    // 从DB获取插件配置数据
    List<PluginData> pluginDataList = pluginService.listAll();
    // 发布插件配置事件
    eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, type, pluginDataList));
    // 从DB获取选择器配置数据
    List<SelectorData> selectorDataList = selectorService.listAll();
    // 发布选择器配置事件
    eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, type, selectorDataList));
    // 从DB获取规则配置数据
    List<RuleData> ruleDataList = ruleService.listAll();
    // 发布规则事件
    eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.RULE, type, ruleDataList));
    // 从DB获取元数据配置数据，并发布事件
    metaDataService.syncData();
    return true;
}

当配置数据事件发布成功后，需有Listener去监听这些事件并继续处理。

DataChangedEventDispatcher

事件分发器，负责将不同的事件由不同的具体的Listener去处理。

public void onApplicationEvent(final DataChangedEvent event) {
    // 监听事件的Listener列表，如websocket、zookeeper、nacos、http等
    for (DataChangedListener listener : listeners) {
        // 根据事件的group判断不同的配置数据
        switch (event.getGroupKey()) {
            case APP_AUTH:  // 认证授权
                // ..
            case PLUGIN: // 插件
                listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
                break;
            case RULE: // 规则
            case SELECTOR: // 选择器
            case META_DATA: // 元数据
            default:
                throw new IllegalStateException("Unexpected value: " + event.getGroupKey());
        }
    }
}

ZookeeperDataChangedListener

负责将配置数据注册到Zookeeper上。

public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
    for (PluginData data : changed) {
        // 创建 plugin's zookeeper path
        final String pluginPath = ZkPathConstants.buildPluginPath(data.getName());
        // delete 如果事件类型是删除，将数据删除
        if (eventType == DataEventTypeEnum.DELETE) {
            // 删除插件Node
            deleteZkPathRecursive(pluginPath);
            final String selectorParentPath = ZkPathConstants.buildSelectorParentPath(data.getName());
            // 删除该插件关联的选择器Node列表
            deleteZkPathRecursive(selectorParentPath);
            final String ruleParentPath = ZkPathConstants.buildRuleParentPath(data.getName());
            // 删除该插件关联的规则Node列表
            deleteZkPathRecursive(ruleParentPath);
            continue;
        }
        // 如果事件类型是其它则创建或更新Zookeeper Node数据
        upsertZkNode(pluginPath, data);
    }
}

private void upsertZkNode(final String path, final Object data) {
    // 如果path 不存在，则先创建path
    if (!zkClient.exists(path)) {
        zkClient.createPersistent(path, true);
    }
    // 写入数据
    zkClient.writeData(path, data);
}

// 删除path, 包插子节点的path都删除
private void deleteZkPathRecursive(final String path) { 
    if (zkClient.exists(path)) {
        zkClient.deleteRecursive(path);
    }
}

Soul网关

监听Zookeeper上的配置数据是否有变化，如果有变化那么Zookeeper就会把变更后的数据push到Soul网关本地内存中。

核心类

ZookeeperSyncDataService

功能：

向Zookeeper注册Listener
将数据同步到本地内存中

// 监听插件、选择器、规则相关的配置数据
private void watcherData() {
    // plugin root path
    final String pluginParent = ZkPathConstants.PLUGIN_PARENT;
    // 所有plugin的paths
    List<String> pluginZKs = zkClientGetChildren(pluginParent);
    for (String pluginName : pluginZKs) {
        // 注册plugins, selectors和rules相关的Listener
        watcherAll(pluginName);
    }
    // 向zookeeper注册 IZKChildListener。一量plugin children path有变更就通知网关
    zkClient.subscribeChildChanges(pluginParent, (parentPath, currentChildren) -> {
        if (CollectionUtils.isNotEmpty(currentChildren)) {
            for (String pluginName : currentChildren) {
                watcherAll(pluginName);
            }
        }
    });
}

private void watcherAll(final String pluginName) {
    // 监听插件配置
    watcherPlugin(pluginName);
    // 监听选择器配置
    watcherSelector(pluginName);
    // 监听规则器配置
    watcherRule(pluginName);
}

private void watcherPlugin(final String pluginName) {
    // 指定的plugin path
    String pluginPath = ZkPathConstants.buildPluginPath(pluginName);
    // 如path 不存在则创建一个path
    if (!zkClient.exists(pluginPath)) {
        zkClient.createPersistent(pluginPath, true);
    }
    // 从zookeeper中获取插件配置数据并写入本地内存中
    cachePluginData(zkClient.readData(pluginPath));
    // 监听该插件的数据变化
    subscribePluginDataChanges(pluginPath, pluginName);
}

private void subscribePluginDataChanges(final String pluginPath, final String pluginName) {
    // 注册 IZkDataListener，用来监听数据变化
    zkClient.subscribeDataChanges(pluginPath, new IZkDataListener() {

        // 数据发生更新
        @Override
        public void handleDataChange(final String dataPath, final Object data) {
            // pluginDataSubscriber 用来更新本地内存
            Optional.ofNullable(data)
                    .ifPresent(d -> Optional.ofNullable(pluginDataSubscriber).ifPresent(e -> e.onSubscribe((PluginData) d)));
        }

        // 数据删除
        @Override
        public void handleDataDeleted(final String dataPath) {
            final PluginData data = new PluginData();
            data.setName(pluginName);
            // pluginDataSubscriber 用来更新本地内存
            Optional.ofNullable(pluginDataSubscriber).ifPresent(e -> e.unSubscribe(data));
        }
    });
}

Zookeeper

配置数据在Zookeeper上的结构如下图：

总结

了解Soul网关是如何通过Zookeeper进行数据同步。