Soul網關學習Zookeeper數據同步02

Soul網關學習Zookeeper數據同步02

做者: 李權

啓動admin，與網關。 admin操做，使用zookeeper同步數據到網關

上一篇，經過soul-admin啓動過程爲入口，分析了soul-admin 啓動就會同步網關數據 rule、metaData、selector、plugin 等到 zookeeper。

數據變化會發布 DataChangedEvent事件，監聽事件將數據同步至zookeeper。 本篇接着上一篇繼續跟蹤源碼分析zookeeper同步數據到網關原理：

• soul-admin 變動網關數據，跟蹤數據同步過程。
• soul-bootstrap 如何獲取zookeeper數據的，如何感知網關數據變化的。

1、soul-admin 變動網關數據，跟蹤數據同步過程

一、在網關後臺嘗試更改divide插件狀態，debug跟蹤。

二、插件更新後會發佈一個DataChangedEvent事件

三、org.dromara.soul.admin.listener.DataChangedEventDispatcher --> onApplicationEvent() 負責監聽事件

四、org.dromara.soul.admin.listener.zookeeper.ZookeeperDataChangedListener 負責同步數據至zookeeper

2、soul-bootstrap 如何獲取zookeeper數據的，如何感知網關數據變化的。

一、soul-bootstrap 依賴

<dependency>
    <groupId>org.dromara</groupId>
    <artifactId>soul-spring-boot-starter-sync-data-zookeeper</artifactId>
    <version>${project.version}</version>
</dependency>
複製代碼

二、soul-bootstrap 啓動後會自動注入org.dromara.soul.spring.boot.sync.data.zookeeper.ZookeeperSyncDataConfiguration

讀取Zookeeper配置向容器中注入ZkClient。

SyncDataService 向容器注入數據同步服務bean，從Spring容器中獲取，ZkClient（zookeeper客戶端）， pluginSubscriber（插件數據訂閱）、metaSubscribers （元數據訂閱）、authSubscribers（權限訂閱）。

public class ZookeeperSyncDataConfiguration {
    /** * Sync data service sync data service. * @param zkClient the zk client * @param pluginSubscriber the plugin subscriber * @param metaSubscribers the meta subscribers * @param authSubscribers the auth subscribers * @return the sync data service */
    @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));
    }
    /** * register zkClient in spring ioc. * @param zookeeperConfig the zookeeper configuration * @return ZkClient {@linkplain ZkClient} */
    @Bean
    public ZkClient zkClient(final ZookeeperConfig zookeeperConfig) {
        return new ZkClient(zookeeperConfig.getUrl(), zookeeperConfig.getSessionTimeout(), zookeeperConfig.getConnectionTimeout());
    }
}
複製代碼

三、org.dromara.soul.sync.data.zookeeper.ZookeeperSyncDataService 初始化，也就是soul-bootstrap啓動後就會從zookeeper獲取數據，同步至內存。

• watcherData()--> watcherAll() --> watcherPlugin() --> cachePluginData()。
• zkClient.subscribeDataChanges() 監聽 當前節點和子節點的內容修改、刪除。

public class ZookeeperSyncDataService implements SyncDataService, AutoCloseable {
    private final ZkClient zkClient;
    private final PluginDataSubscriber pluginDataSubscriber;
    private final List<MetaDataSubscriber> metaDataSubscribers;
    private final List<AuthDataSubscriber> authDataSubscribers;
    /** * Instantiates a new Zookeeper cache manager. * @param zkClient the zk client * @param pluginDataSubscriber the plugin data subscriber * @param metaDataSubscribers the meta data subscribers * @param authDataSubscribers the auth data subscribers */
    public ZookeeperSyncDataService(final ZkClient zkClient, final PluginDataSubscriber pluginDataSubscriber, final List<MetaDataSubscriber> metaDataSubscribers, final List<AuthDataSubscriber> authDataSubscribers) {
        this.zkClient = zkClient;
        this.pluginDataSubscriber = pluginDataSubscriber;
        this.metaDataSubscribers = metaDataSubscribers;
        this.authDataSubscribers = authDataSubscribers;
        watcherData();
        watchAppAuth();
        watchMetaData();
    }
    ......
	private void watcherData() {
	    final String pluginParent = ZkPathConstants.PLUGIN_PARENT;
	    List<String> pluginZKs = zkClientGetChildren(pluginParent);
	    for (String pluginName : pluginZKs) {
	        watcherAll(pluginName);
	    }
	    zkClient.subscribeChildChanges(pluginParent, (parentPath, currentChildren) -> {
	        if (CollectionUtils.isNotEmpty(currentChildren)) {
	            for (String pluginName : currentChildren) {
	                watcherAll(pluginName);
	            }
	        }
	    });
	}
    ......
	private void watcherPlugin(final String pluginName) {
	    String pluginPath = ZkPathConstants.buildPluginPath(pluginName);
	    if (!zkClient.exists(pluginPath)) {
	        zkClient.createPersistent(pluginPath, true);
	    }
	    cachePluginData(zkClient.readData(pluginPath));
	    subscribePluginDataChanges(pluginPath, pluginName);
	}
}
複製代碼

四、debug過程

3、soul-bootstrap 是如何感知網關數據變化的

一、org.dromara.soul.sync.data.zookeeper.ZookeeperSyncDataService cacheRuleData 方法上打上斷點，更新插件規則，觀察是否會進入此斷點。

private void cacheRuleData(final RuleData ruleData) {
    Optional.ofNullable(ruleData)
            .ifPresent(data -> Optional.ofNullable(pluginDataSubscriber).ifPresent(e -> e.onRuleSubscribe(data)));
}
複製代碼

二、soul-admin後臺操做更改divide插件規則，首先soul-admin會發布事件，並監聽事件同步更新數據至zookeeper。

三、soul-bootstrap 確實收到了插件數據的更新，根據Soul官網介紹的"zookeeper 的同步原理"這裏主要是依賴 zookeeper 的 watch 機制。

org.dromara.soul.sync.data.zookeeper.ZookeeperSyncDataService 類：

zkClient.subscribeDataChanges() 監聽 當前節點和子節點的內容修改、刪除。

zkClient.subscribeChildChanges(groupParentPath, (parentPath, currentChildren) -> {
    if (CollectionUtils.isNotEmpty(currentChildren)) {
        List<String> addSubscribePath = addSubscribePath(childrenList, currentChildren);
        // Get the newly added node data and subscribe to that node
        addSubscribePath.stream().map(addPath -> {
            String realPath = buildRealPath(parentPath, addPath);
            cacheRuleData(zkClient.readData(realPath));
            return realPath;
        }).forEach(this::subscribeRuleDataChanges);
    }
});
private void subscribeRuleDataChanges(final String path) {
    zkClient.subscribeDataChanges(path, new IZkDataListener() {
        @Override
        public void handleDataChange(final String dataPath, final Object data) {
            cacheRuleData((RuleData) data);
        }
        @Override
        public void handleDataDeleted(final String dataPath) {
            unCacheRuleData(dataPath);
        }
    });
}
複製代碼

4、總結