Soul网关源码分析-11期

后台与网关数据同步 (Zookeeper篇)

---

后台与网关的数据同步, 在 V2.2.1 中默认是 Websocket 方式, 如何切换到 Zookeeper 呢?

这里肯定是后台与网关都切换到 Zookeeper, 先分析第一个如何切换后台的信息同步模式.

后台信息模式切换

没有什么思路的情况下, 可以先从旧有模式入手, 知道旧有模式 (Websocket) 的启动方式, 即可了解其他方式的启动.

在上个文章 (Soul网关源码分析-10期) 中有总结到, 后台通过 DataSyncConfiguration 配置类去注入 Websocket 监听器为 Spring Bean, 相当于是启动类了.

@Configuration
public class DataSyncConfiguration {@Configuration@ConditionalOnProperty(prefix = "soul.sync.zookeeper", name = "url")@Import(ZookeeperConfiguration.class)static class ZookeeperListener {@Bean@ConditionalOnMissingBean(ZookeeperDataChangedListener.class)public DataChangedListener zookeeperDataChangedListener(final ZkClient zkClient) {return new ZookeeperDataChangedListener(zkClient);}@Bean@ConditionalOnMissingBean(ZookeeperDataInit.class)public ZookeeperDataInit zookeeperDataInit(final ZkClient zkClient, final SyncDataService syncDataService) {return new ZookeeperDataInit(zkClient, syncDataService);}}@Configuration@ConditionalOnProperty(name = "soul.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)@EnableConfigurationProperties(WebsocketSyncProperties.class)static class WebsocketListener {@Bean@ConditionalOnMissingBean(WebsocketDataChangedListener.class)public DataChangedListener websocketDataChangedListener() {return new WebsocketDataChangedListener();}@Bean@ConditionalOnMissingBean(WebsocketCollector.class)public WebsocketCollector websocketCollector() {return new WebsocketCollector();}@Bean@ConditionalOnMissingBean(ServerEndpointExporter.class)public ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}}
}

看到这就知道该怎么做了, DataSyncConfiguration 配置类根据配置文件的参数, 决定各个监听器 (WebsocketListener、ZookeeperListener) 是否使用.

我们在配置文件中将 “soul.sync.websocket.enabled” 的参数改为 “false”, 并将 “soul.sync.zookeeper.url” 的值写入, 来启用后台 Zookeeper 信息同步. 这段的配置如下 (yml格式):

soul:sync:websocket:enabled : falsezookeeper:url: localhost:2181sessionTimeout: 5000connectionTimeout: 2000

用可视化工具看看 Zookeeper 里的信息, 确认后台将元数据信息都写到 Zookeeper 上了.

网关信息模式切换

后台模式切换完成, 接下来就是网关, 这块我们也根据刚刚的经验, 先找到旧模式的启动类.

依旧根据之前的文章分析, 找到启动类 WebsocketSyncDataConfiguration

@Configuration
@ConditionalOnClass(WebsocketSyncDataService.class)
@ConditionalOnProperty(prefix = "soul.sync.websocket", name = "urls")
@Slf4j
public class WebsocketSyncDataConfiguration {// ...
}

这里只能得到一个信息, 就是如何开启or关闭 Websocket 通信方式, 但不知道怎么开启 Zookeeper 方式. 别急我们看下这个类的所属项目, 如果项目结构设计的好, 其他同步模式的专门做 Spring Bean 注入的启动项目, 也应该和这个项目类似名称.

找到 Zookeeper 模式的启动项 soul-spring-boot-starter-sync-data-zookeeper , 顺势找到启动配置类 ZookeeperSyncDataConfiguration, 看看什么样的参数能开启它.

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

最后, 确认网关的 pom 中已引入 soul-spring-boot-starter-sync-data-zookeeper , 并按如下修改配置信息:

soul:sync:
#		 websocket:
#			 urls: ws://localhost:9095/websocketzookeeper:url: localhost:2181sessionTimeout: 5000connectionTimeout: 2000

后台数据初始化时传输

搞懂如何切换后, 现在来分析两个问题: 后台怎么将元数据传输给 Zookeeper, 以及网关接收到数据变更后怎么做.

后台数据初始化来自与项目容器启动, 找到 DataSyncConfiguration 配置 Bean 时的关键类 ZookeeperDataInit

public class ZookeeperDataInit implements CommandLineRunner {@Overridepublic void run(final String... args) {String pluginPath = ZkPathConstants.PLUGIN_PARENT;String authPath = ZkPathConstants.APP_AUTH_PARENT;String metaDataPath = ZkPathConstants.META_DATA;if (!zkClient.exists(pluginPath) && !zkClient.exists(authPath) && !zkClient.exists(metaDataPath)) {// 这个地方会同步所有信息给 ZookeepersyncDataService.syncAll(DataEventTypeEnum.REFRESH);}}
}

找到 SyncDataServiceImpl#syncAll 方法, 可以看到发送了五种类型数据事件.

@Service("syncDataService")
public class SyncDataServiceImpl implements SyncDataService {@Overridepublic boolean syncAll(final DataEventTypeEnum type) {// 事件类型: appAuth appAuthService.syncData();List<PluginData> pluginDataList = pluginService.listAll();// 事件类型: plugineventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, type, pluginDataList));List<SelectorData> selectorDataList = selectorService.listAll();// 事件类型: selectoreventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, type, selectorDataList));List<RuleData> ruleDataList = ruleService.listAll();// eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.RULE, type, ruleDataList));metaDataService.syncData();return true;}
}

最终这些事件会经过实现了 Spring 发布订阅模式的事件分发器 DataChangedEventDispatcher , 流向 ZookeeperDataChangedListener

public class ZookeeperDataChangedListener implements DataChangedListener {@Overridepublic void onAppAuthChanged(final List<AppAuthData> changed, final DataEventTypeEnum eventType) {// ...}@Overridepublic void onMetaDataChanged(final List<MetaData> changed, final DataEventTypeEnum eventType) {for (MetaData data : changed) {// ...// 写入数据到 ZookeeperzkClient.writeData(metaDataPath, data);}}@Overridepublic void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {// ...}@Overridepublic void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {// ...}@Overridepublic void onRuleChanged(final List<RuleData> changed, final DataEventTypeEnum eventType) {// ...}
}

后台数据变动时传输

后台数据变动来自与网页端, 而后台的 Zookeeper 监听器在变动时肯定会被触发, 并写入数据到 Zookeeper中, 我们找到后台启动时被初始化的监听器 DataSyncConfiguration, debug它的调用链

这里多个来源与 DataChangedListener 的抽象方法, 区分了前端的 “同步数据” 的元数据类型, 追溯调用最终找到各个元数据类型所对应的 Controller 类

@RestController
@RequestMapping("/meta-data")
public class MetaDataController {@PostMapping("/syncData")public SoulAdminResult syncData() {metaDataService.syncData();return SoulAdminResult.success();}
}

在具体的插件页做 “同步数据” 有些许不同.

比如在网页的 [插件列表]-[divide] 这里点击 “同步divide” , 会通过 PluginController 以及 SyncDataServiceImpl#syncPluginData, 走到 Zookeeper 监听器这里的三个方法, 触发 selector选择器、plugin插件、rule规则的数据变更调用.

PS: 在[系统管理]-[插件管理] 中做同步操作也会触发三个不同类型事件, 仅是 Controller 方法不同

@RestController
@RequestMapping("/plugin")
public class PluginController {@PutMapping("/syncPluginData/{id}")public SoulAdminResult syncPluginData(@PathVariable("id") final String id) {boolean success = syncDataService.syncPluginData(id);if (success) {return SoulAdminResult.success("sync plugins success");} else {return SoulAdminResult.error("sync plugins fail");}}
}

@Service("syncDataService")
public class SyncDataServiceImpl implements SyncDataService {@Overridepublic boolean syncPluginData(final String pluginId) {PluginVO pluginVO = pluginService.findById(pluginId);// 发送插件信息变更通知eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, DataEventTypeEnum.UPDATE,Collections.singletonList(PluginTransfer.INSTANCE.mapDataTOVO(pluginVO))));List<SelectorData> selectorDataList = selectorService.findByPluginId(pluginId);if (CollectionUtils.isNotEmpty(selectorDataList)) {// 发送选择器信息变更通知eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.REFRESH, selectorDataList));List<RuleData> allRuleDataList = new ArrayList<>();for (SelectorData selectData : selectorDataList) {List<RuleData> ruleDataList = ruleService.findBySelectorId(selectData.getId());allRuleDataList.addAll(ruleDataList);}// 发送规则信息变更通知eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.RULE, DataEventTypeEnum.REFRESH, allRuleDataList));}return true;}
}

网关数据变动时接收

网关端在启动时就开启了 Zookeeper 监听, 利用 Watch 机制监听 Zookeeper 中节点的变动. 具体可看下 ZookeeperSyncDataService 的代码

public class ZookeeperSyncDataService implements SyncDataService, AutoCloseable {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;// 这三个方法即是开启 Zookeeper watchwatcherData();watchAppAuth();watchMetaData();}// ...
}

构造器被调用后, 这个类会去 Zookeeper 中找到约定好的路径下的所有文件, 并监听他们的状态.

有一点疑问, 如果后台比网关后启动, Zookeeper中没有值, 那么网关是否即获取不到任何数据, 也无法监听变化, 后续后台即使启动也没用呢?

看看 ZookeeperSyncDataService#watcherData 的代码, 这里订阅了子节点变化信息, 所以后台如果新增节点, 这块也能触发并 watch 住新节点

private void watcherData() {// ...zkClient.subscribeChildChanges(pluginParent, (parentPath, currentChildren) -> {// 节点变化再次 watchif (CollectionUtils.isNotEmpty(currentChildren)) {for (String pluginName : currentChildren) {watcherAll(pluginName);}}});
}

如果节点数据发生变动, 即会通知这里并调用 ZookeeperSyncDataService 下的各个订阅器

private final PluginDataSubscriber pluginDataSubscriber;private final List<MetaDataSubscriber> metaDataSubscribers;private final List<AuthDataSubscriber> authDataSubscribers;

这些订阅器就会调用各个插件自己的订阅器实现了, 比如 divide 插件项目的 DividePluginDataHandler#handlerSelector, 这块的功能是网关数据扭转到各个扩展插件, 在上一篇中也有分析 Soul网关源码分析-10期, 有兴趣的话可以翻看一二.