本篇主要对 SOFARegistry 的数据同步模块进行解析，对于注册中心的概念以及 SOFARegistry 的基础架构不做过多阐述。
本文主要写作思路大致分为下面 2 个部分：第一部分借助 SOFARegistry 中的角色分类来说明哪些角色之间会进行数据同步，第二部分对数据同步的具体实现进行解析。
如上图，SOFARegistry 包含 4 个角色：
角色 | 说明 |
Client | 提供应用接入服务注册中心的基本 API 能力，应用系统通过依赖客户端 JAR 包，通过编程方式调用服务注册中心的服务订阅和服务发布能力。 |
SessionServer | 会话服务器，负责接受 Client 的服务发布和服务订阅请求，并作为一个中间层将写操作转发 DataServer 层。SessionServer 这一层可随业务机器数的规模的增长而扩容。 |
DataServer | 数据服务器，负责存储具体的服务数据，数据按 dataInfoId 进行一致性 Hash 分片存储，支持多副本备份，保证数据高可用。这一层可随服务数据量的规模的增长而扩容。 |
MetaServer | 元数据服务器，负责维护集群 SessionServer 和 DataServer 的一致列表，作为 SOFARegistry 集群内部的地址发现服务，在 SessionServer 或 DataServer 节点变更时可以通知到整个集群。 |
在这 4 个角色中，MetaServer 作为元数据服务器本身不处理实际的业务数据，仅负责维护集群 SessionServer 和 DataServer 的一致列表，不涉及数据同步问题；Client 与 SessionServer 之间的核心动作是订阅和发布，从广义上来说，属于用户侧客户端与 SOFARegistry 集群的数据同步，可以见：github.com/sofastack/s…，因此不在本文讨论范畴之内。
SessionServer 作为会话服务，它主要解决海量客户端连接问题，其次是缓存客户端发布的所有 pub 数据；session 本身不持久化服务数据，而是将数据转写到 DataServer。DataServer 存储服务数据是按照 dataInfoId 进行一致性 Hash 分片存储的，支持多副本备份，保证数据高可用。
从 SessionServer 和 DataServer 的功能分析中可以得出：
SOFARegistry 中对于上述两个对于数据一致性保证就是通过数据同步机制来实现的。
下面主要介绍数据同步的实现细节，主要包括 SessionServer 和 DataServer 之间的数据同步 和 DataServer 多副本之间的数据同步两块。
SessionServer 和 DataServer 之间的数据同步，是基于推拉结合的机制
关于推和拉两种模式检查的 version 有一些差异，可以详见下面 推模式下的数据同步 和 **拉模式下的数据同步 **中的具体介绍
推模式是通过 SyncingWatchDog 这个守护线程不断 loop 执行来实现数据变更检查和通知发起的。
// 这里遍历所有的 slot
for (SlotState slotState : slotTableStates.slotStates.values()) {
try {
sync(slotState, syncSessionIntervalMs, syncLeaderIntervalMs, slotTableEpoch);
} catch (Throwable e) {
SYNC_ERROR_LOGGER.error(
"[syncCommit]failed to do sync slot {}, migrated={}",
slotState.slot,
slotState.migrated,
e);
}
}按 slot 分组汇总数据版本。data 与每个 session 的连接都对应一个 SyncSessionTask，SyncSessionTask 负责执行同步数据的任务，核心同步逻辑在 com.alipay.sofa.registry.server.data.slot.SlotDiffSyncer#sync方法中完成，大致流程如下面时序图所示：
这上图圈红部分的逻辑第四步，根据 dataInfoId diff 更新 data 内存数据，这里仅处理了被移除的 dataInfoId，对于新增和更新的没有做任务处理，而是通过后面的第 5 -7 步来完成；这么做的主要原因在于避免产生空推送导致一些危险情况发生。
第 5 步中，比较的是所有变更 dataInfoId 的 pub version，具体比较逻辑参考后面 diffPublisher 小节中的介绍。
数据变更事件会被收集在 DataChangeEventCenter 的 dataCenter2Changes 缓存中，然后由一个守护线程 ChangeMerger 负责从 dataCenter2Changes 缓存中不断的读取，这些被捞到的事件源会被组装成 ChangeNotifier 任务，提交给一个单独的线程池(notifyExecutor)处理，整个过程全部是异步的。
拉模式下，由 SessionServer 负责发起，com.alipay.sofa.registry.server.session.registry.SessionRegistry.VersionWatchDog默认情况下每 5 秒扫描一次版本数据，如果版本有发生变更，则主动进行拉取一次，流程大致如下：
需要注意的是，拉模式对推送流程的补充，这里的 version 是每个 sub 的 lastPushedVersion， 而 推模式的version 是 pub 的数据的 version。关于 lastPushedVersion 的获取可以参考 com.alipay.sofa.registry.server.session.store.SessionInterests#selectSubscribers
store.forEach((String dataInfoId, Map subs) -> {
// ...
long maxVersion = 0;
for (Subscriber sub : subs.values()) {
// ...
// 获取当前 sub 的 pushVersion
final long pushVersion = sub.getPushedVersion(dataCenter);
// 如果 pushVersion 比最大(最新)版本大，则将当前 pushVersion 作为最新版本推送版本
if (maxVersion < pushVersion) {
maxVersion = pushVersion;
}
}
versions.put(dataInfoId, new DatumVersion(maxVersion));
}); 主要是 slot对应的 data 的 follower 定期和 leader 进行数据同步，其同步逻辑与 SessionServer 和 DataServer 之间的数据同步逻辑差异不大；发起方式也是一样的；data 判断如果当前节点不是 leader，就会进行与 leader 之间的数据同步。
if (localIsLeader(slot)) {
// 如果当前是 leader，则执行 session 同步或者 migrating
} else {
// 如果当前不是 leader，则和 leader 同步数据
syncLeader(slotState, syncLeaderIntervalMs, slotTableEpoch);
}篇幅原因，这部分不展开讨论。
不管是 SessionServer 和 DataServer 之间的同步，还是 DataServer 多副本之间的同步，都是基于增量 diff 同步的，不会一次性同步全量数据。本节对增量同步 diff 计算逻辑进行简单分析，核心代码在 com.alipay.sofa.registry.common.model.slot.DataSlotDiffUtils（建议阅读这部分代码时直接结合代码中的测试用例来看）；主要包括计算 digest 和 publishers 两个。
DataSlotDiffUtils#diffDigest 方法接收两个参数
核心计算逻辑如下代码分析
// 遍历 sourceDigestMap 元素
for (Map.Entry e : sourceDigestMap.entrySet()) {
// dataInfoId
final String dataInfoId = e.getKey();
// 从 目标数据 集中根据 dataInfoId 获取数据摘要
DatumDigest targetDigest = targetDigestMap.get(dataInfoId);
// 如果目标数据集中没有当前 dataInfoId 对应的数据摘要，
// 则将当前 dataInfoId 作为新增项
if (targetDigest == null) {
adds.add(dataInfoId);
continue;
}
// 如果目标数据集中有当前 dataInfoId 对应的数据摘要，
// 但是数据摘要不同，则将当前 dataInfoId 作为待更新项
if (!targetDigest.equals(e.getValue())) {
updates.add(dataInfoId);
}
}
// 如果目标数据集中的 dataInfoId 不再基线数据集中时，
// 则将当前 dataInfoId 作为待移除项。
List removes = new ArrayList<>();
for (String dataInfoId : targetDigestMap.keySet()) {
if (!sourceDigestMap.containsKey(dataInfoId)) {
removes.add(dataInfoId);
}
} 那么根据上述 diff 计算逻辑，这里有如下几种场景（假设基线数据集数据中 dataInfoId 为 a 和 b）
diffPublisher 与 diffDigest 计算稍有不同，diffPublisher 接收三个参数，除了目标数据集和基线数据集之外，还有一个 publisherMaxNum（默认 400），用于限制每次处理的数据个数；这里同样给出核心代码解释：
// 遍历所有目标数据集
for (DatumSummary summary : targetDatumSummaries) {
// 拿到 dataInfoId
final String dataInfoId = summary.getDataInfoId();
// 看基线数据集中是否包括当前 dataInfoId 对应的 Publisher 数据
Map publisherMap = sourcePublishers.get(dataInfoId);
// 这里表示 dataInfoId 移除被移除了，不需要做任何处理
if (publisherMap == null) { continue; }
Set registerIds = summary.getPublisherVersions().keySet();
// 遍历 registerIds
for (String registerId : registerIds) {
// 如果基线数据集中不包括此 registerId，则将当前 registerId 加入待移除列表中
if (!publisherMap.containsKey(registerId)) {
List list = removedPublishers.computeIfAbsent(dataInfoId, k -> new ArrayList<>());
list.add(registerId);
}
}
List publishers = new ArrayList<>();
Map versions = summary.getPublisherVersions();
// 遍历版本
for (Map.Entry p : publisherMap.entrySet()) {
final String registerId = p.getKey();
// 如果目标数据集当前 dataInfoId 的 registerIds 集中不包括基线的
// 则作为更新项
if (!versions.containsKey(registerId)) {
publishers.add(p.getValue());
continue;
}
// 如果当前 registerId 版本相同，则不做处理
if (p.getValue().registerVersion().equals(versions.get(registerId))) {
// the same
continue;
}
// 不相等，则作为更新项
publishers.add(p.getValue());
}
} 这里同样分析几种场景（下面只的是更新 dataInfoId 对应的 publisher，registerId 与 publisher是 一一对应）：
本文主要介绍了 SOFARegistry 中数据同步模块；首先从 SOFARegistry 角色分类阐述不同角色之间存在的数据同步问题，针对其中 SessionServer 与 DataServer 数据同步 和 DataServer 多副本之间数据同步进行了展开分析；在 SessionServer 与 DataServer 数据同步分析中，着重分析了推和拉两种场景下数据同步的整体流程；最后对 SOFARegistry 中数据增加的 diff 计算逻辑进行了介绍，并结合相关核心代码描述了具体的场景。
整体来看，SOFARegistry 数据同步上的处理上有一些点值得我们学习：
作者：磊叔的技术博客
链接：https://juejin.cn/post/7116820785217404941
来源：稀土掘金
| 留言与评论（共有 0 条评论） “” |
