soul 的数据同步也支持使用 nacos，这个我之前没用过，甚至没听过，网上查了一遍，只知道是阿里巴巴开源的，一个更易于构建云原生应用的动态服务发现、配置管理和服务管理平台，nacos 可作为微服务核心的服务注册与发现中心，这个看完一圈还是不理解。上一篇讲到分布式系统的 CAP 理论，看到说 nacos 同时支持 CP 和 AP 原则，支持切换，我觉得应该是比 zookeeper 更强一些，也是用来存储数据的。
使用 nacos 做数据同步，在 application-local.yml 配置 nacos
soul: sync: nacos: url: localhost:8848 namespace: 1c10d748-af86-43b9-8265-75f487d20c6c acm: enabled: false endpoint: acm.aliyun.com namespace: accessKey: secretKey:
复制代码
当数据变更时，是使用 NacosDataChangedListener 来处理的。
@Override public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) { //getConfig 是从 configService 读取配置 updatePluginMap(getConfig(PLUGIN_DATA_ID)); switch (eventType) { case DELETE: changed.forEach(plugin -> PLUGIN_MAP.remove(plugin.getName())); break; case REFRESH: case MYSELF: Set<String> set = new HashSet<>(PLUGIN_MAP.keySet()); changed.forEach(plugin -> { set.remove(plugin.getName()); PLUGIN_MAP.put(plugin.getName(), plugin); }); PLUGIN_MAP.keySet().removeAll(set); break; default: changed.forEach(plugin -> PLUGIN_MAP.put(plugin.getName(), plugin)); break; } //Map 转为 json，把最新数据发布出去。 publishConfig(PLUGIN_DATA_ID, PLUGIN_MAP); }
//todo 这一块代码没看太懂，留个坑，明天接着看 private void updatePluginMap(final String configInfo) { JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class); Set<String> set = new HashSet<>(PLUGIN_MAP.keySet()); for (Entry<String, JsonElement> e : jo.entrySet()) { set.remove(e.getKey()); PLUGIN_MAP.put(e.getKey(), GsonUtils.getInstance().fromJson(e.getValue(), PluginData.class)); } PLUGIN_MAP.keySet().removeAll(set); }
复制代码
soul-bootstrap 启动时，使用 soul-spring-boot-starter-sync-data-nacos 包下的 NacosSyncDataConfiguration ，返回了一个 NacosSyncDataService 实例，这个类继承了 NacosCacheHandler，很多处理逻辑都是在这个类里完成的。
protected void updatePluginMap(final String configInfo) { try { List<PluginData> pluginDataList = GsonUtils.getInstance().toObjectMap(configInfo, PluginData.class).values().stream().collect(Collectors.toList()); pluginDataList.forEach(pluginData -> Optional.ofNullable(pluginDataSubscriber).ifPresent(subscriber -> { //删除之前的数据 subscriber.unSubscribe(pluginData); subscriber.onSubscribe(pluginData); })); } catch (JsonParseException e) { log.error("sync plugin data have error:", e); } }
复制代码
这块代码，可以看出来，nacosv 实现的是全量更新，它先删除之前缓存里的数据，再把最新的数据放到缓存里，而之前分析的 zookeeper，websocket 实现的都是增量更新，只更新一小部分。
今天虽然走了一遍流程，但具体细节部分还很多不理解，明天再继续分析。
评论