源码解析：无损运维
SOFARegistry 是一个基于内存存储的分布式注册中心，数据是分散存储在各个节点，为了做到注册中心自身运维期间依然能够对外正常提供服务，需要进行节点下线快速感知和数据迁移。
session 存储 client 发送的发布和订阅数据，由于 client 断连重放的特性，session 单机下线后 client 会重放数据到其他节点。
data 会接收 session 发送的发布数据，data 单机下线后，meta 会通过 SlotTable 的变更把对应 Slot 的所有权移交其他 data，data 启动后会进行数据同步到达数据完整可对外提供服务的状态。
session、data下线的过程，如何避免下线期间数据丢失和抖动
SessionServer 和 DataServer 下线的相关代码都位于各自 bootstrap 类的 dostop 方法中。
首先讨论 SessionServer 的下线过程：
private void doStop() { try { LOGGER.info("{} Shutting down Session Server..", new Date().toString()); stopHttpServer(); clientNodeConnectionHandler.stop(); // stop process disconnect event stopServer(); // stop http server and client bolt server before add blacklist // make sure client reconnect to other sessions and data gracefulShutdown(); stopDataSyncServer(); stopConsoleServer(); executorManager.stopScheduler(); } catch (Throwable e) { LOGGER.error("Shutting down Session Server error!", e); } LOGGER.info("{} Session server is now shutdown...", new Date().toString()); } 从代码中可以看出，SessionServer 服务在下线时，首先会停掉 HttpServer，HttpServer 提供一系列 REST 接口，用于 dashboard 管理、数据查询等；然后停掉 clientNodeConnectionHandler，clientNodeConnectionHandler 是 Exchange 设置的一系列 ChannelHandler 之一，Exchange 作为 Client / Server 连接的抽象，负责节点之间的连接。gracefulShutdown 方法则是通知 Meta（元数据服务器）将本节点加入 Meta 的黑名单中。HttpServer 和客户端的 bolt Server 是在本节点加入黑名单前关停，以保证 client 已经重连到了其他 session 节点上。这样就保证了运维的 session 节点下线期间，client 的数据不会因为 session 节点的不可用，导致数据丢失和抖动。接下来又依次关停了 DataSyncServer，ConsoleServer。
DataServer 下线过程的代码如下：
private void doStop() { try { LOGGER.info("{} Shutting down Data Server..", new Date().toString()); gracefulShutdown(); stopHttpServer(); stopServer(); stopDataSyncServer(); stopNotifyServer(); } catch (Throwable e) { LOGGER.error("Shutting down Data Server error!", e); } LOGGER.info("{} Data server is now shutdown...", new Date().toString()); } DataServer 的下线则和 Session 的下线有些区别，由于 DataServer 数据服务器，负责存储具体的服务数据，而且 Slot 均匀地分配给每个节点上，所以下线前需要检测 DataServer 上的插槽状态，所以 doStop 的方法中，首先调用了如下的优雅下线的方法，其中代码中的主要内容如下：
addBlacklistRetryer.call( () -> { LOGGER.info("[GracefulShutdown] add self to blacklist"); metaServerService.addSelfToMetaBlacklist(); return true; }); addBlacklistRetryer.call( () -> { if (fetchStopPushService.isStopPushSwitch()) { return true; } SlotTableStatusResponse statusResponse = metaServerService.getSlotTableStatus(); if (statusResponse.isProtectionMode()) { return true; } LOGGER.info("[GracefulShutdown] wait no slot"); if (slotManager.hasSlot()) { throw new RuntimeException("current data server still own slot, waiting..."); } return true; }); LOGGER.info("add data self to blacklist successfully"); 首先节点通知 Meta 加入黑名单，stopmetaServerService.getSlotTableStatus() 获取节点上 SlotTable 的状态，当 Slot 重新分配给其他节点后，该 Data 节点才会成功加入黑名单，并进行接下来 HttpServer、DataSyncServer、NotifyServer 的下线动作。
整体流程是 Data 上的 Slot 逐步迁移，期间仍然对外提供服务。
迁移完成后主动从列表中剔除并设置节点黑名单并下线。
以下图为例，假设要下线的节点是 DataServer-1：将 Data-server 添加到注册中心迁移名单中；再将 Slot-2 对应的 follower（DataServer-2）提升为 Leader；等待 Data 节点达到稳定；继续将 Slot-3 对应的 follower（DataServer-3) 提升为 Leader；按照以上步骤分配掉剩余的 follower Slots；Slot 迁移完毕，Data-server 从 Data 列表中剔除，同时添加为黑名单；从迁移名单中设置为 succeed。
Session 和 Data下线中的优雅关闭和数据迁移保证了下线期间数据丢失和抖动，期间仍然对外提供服务。
Meta 如何处理 Session、Data 主动下线的请求通知
Meta 集群服务于 SOFARegistry 内部的 Session 集群和 Data 集群，Meta 层能够感知到 Session 节点和 Data 节点的变化，并通知集群的其它节点。RegistryForbiddenServerHandler 注册在sessionServerHandlers、dataServerHandlers ：
@Bean(name = "sessionServerHandlers") public Collection<AbstractServerHandler> sessionServerHandlers() { Collection<AbstractServerHandler> list = new ArrayList<>(); list.add(heartbeatRequestHandler()); list.add(fetchProvideDataRequestHandler()); list.add(fetchSystemPropertyRequestHandler()); list.add(registryForbiddenServerHandler()); list.add(getSlotTableStatusRequestHandler()); return list; } @Bean(name = "dataServerHandlers") public Collection<AbstractServerHandler> dataServerHandlers() { Collection<AbstractServerHandler> list = new ArrayList<>(); list.add(heartbeatRequestHandler()); list.add(fetchProvideDataRequestHandler()); list.add(registryForbiddenServerHandler()); list.add(fetchSystemPropertyRequestHandler()); list.add(getSlotTableStatusRequestHandler()); return list; } 当有 Data 和 Session 上下线时，会触发 registryForbiddenServerHandler，具体的 doHandle 方法将下线的节点添加到黑名单中：
boolean success = false; switch (operation) { case ADD: success = registryForbiddenServerManager.addToBlacklist(ip); break; case REMOVE: success = registryForbiddenServerManager.removeFromBlacklist(ip); break; default: break; } Meta 的心跳检测 HeartbeatRequestHandler 会在心跳检测时，获取 Session 的状态，
@Override public VersionedList<SessionNode> getSessionServerMetaInfo() { VersionedList<Lease<SessionNode>> leaseMetaInfo = getLeaseMeta(); List<SessionNode> sessionNodes = Lists.newArrayList(); leaseMetaInfo .getClusterMembers() .forEach( lease -> { sessionNodes.add(lease.getRenewal()); }); return new VersionedList<>(leaseMetaInfo.getEpoch(), sessionNodes); } 获取 Meta 租约中最新的 Session 节点的信息，更新 Session 节点的信息。
Data 的节点信息也会通过类似的过程更新到 Meta 中，MetaServerManager 会根据节点类型，同步节点的在线信息：
@Override public VersionedList<DataNode> getDataServerMetaInfo() { VersionedList<Lease<DataNode>> leaseMetaInfo = getLeaseMeta(); List<DataNode> dataNodes = Lists.newArrayList(); leaseMetaInfo .getClusterMembers() .forEach( lease -> { dataNodes.add(lease.getRenewal()); }); return new VersionedList<>(leaseMetaInfo.getEpoch(), dataNodes); } client 重连回放机制
public void ensureConnected() throws InterruptedException { if (isConnected()) { return; } while (!connect()) { Thread.sleep(ClientConnection.RECONNECTING_DELAY); } } 在 client 的工作线程中，会通过上述代码以 5s 为间隔时间，反复检查链接状态，client 的检查函数位于工作线程的一个无限循环的检查语句中:
/** Handle. */ @Override public void handle() { //noinspection InfiniteLoopStatement while (true) { try { // check connection status, try to reconnect to the server when connection lose client.ensureConnected(); if (requestQueue.isEmpty()) { await(config.getRecheckInterval()); continue; } Iterator<TaskEvent> lt = requestQueue.iterator(); while (lt.hasNext()) { client.ensureConnected(); TaskEvent ev = lt.next(); lt.remove(); /**.........**/ Data 启动期以及 Data 的启动过程的状态转化，如何确保新分配的 Slot 的数据完整性
启动入口
Data 启动的过程代码位于com.alipay.sofa.registry.server.data.bootstrap.DataServerBootstrap#start
/** start dataserver */ public void start() { try { LOGGER.info("begin start server"); LOGGER.info("release properties: {}", ServerEnv.getReleaseProps()); LOGGER.info("the configuration items are as follows: " + dataServerConfig.toString()); ReporterUtils.enablePrometheusDefaultExports(); openDataServer(); openDataSyncServer(); openHttpServer(); renewNode(); fetchProviderData(); systemPropertyProcessorManager.startFetchMetaSystemProperty(); startScheduler(); TaskMetrics.getInstance().registerBolt(); postStart(); Runtime.getRuntime().addShutdownHook(new Thread(this::doStop)); LOGGER.info("start server success"); } catch (Throwable e) { throw new RuntimeException("start server error", e); } } DataServer 模块的各个 bean 在 JavaConfig 中统一配置，JavaConfig 类为 DataServerBeanConfiguration， 启动入口类为 DataServerInitializer，该类不由 JavaConfig 管理配置，而是继承了 SmartLifecycle 接口，在启动时由 Spring 框架调用其 start 方法。
该方法中调用了 DataServerBootstrap#start 方法（如上述代码所示），用于启动一系列的初始化服务。
sofaRegistry V6 版本引入了“路由表”（SlotTabel）的概念，“路由表”负责存放每个节点和 N 个 Slot 的映射关系，并保证尽量把所有 Slot 均匀地分配给每个节点。这样，当节点上下线时，只需要修改路由表内容即可。
从代码中可以看出，DataServer 服务在启动时，会启动 DataServer、DataSyncServer、HttpServer 三个 bolt 服务。在启动这些 Server 之时，DataServer 注册了一系列 Handler 来处理各类消息。启动这些 Server 后，调用了renewNode（）方法，renewNode 调用 metaServerService的 renewNode，通过获取在心跳请求中将返回 SlotTable 路由表信息、Data 节点将路由表 SlotTable 保存在本地中，具体代码可以参考com.alipay.sofa.registry.server.meta.remoting.meta.MetaServerRenewService#renewNode，下一步又刷新了 Session 的信息，并创建了一个 renew meta 的守护进程，用来持续刷新节点上存储的集群信息：
private void renewNode() { metaServerService.renewNode(); // init session lease with first renew for (ProcessId processId : metaServerService.getSessionProcessIds()) { sessionLeaseManager.renewSession(processId); } metaServerService.startRenewer(); } slotManagerImpl 是 DataServer 的一个单例 ConditionalOnMissingBean 对象。当 Data 启动时，SlotManage 负责 Slot 数据的更新和迁移（com.alipay.sofa.registry.server.data.slot.SlotManagerImpl#syncMigrating）：
private void syncMigrating( SlotState slotState, Collection<String> sessions, int syncSessionIntervalMs, long slotTableEpoch) { final Slot slot = slotState.slot; if (slotState.migratingStartTime == 0) { slotState.migratingStartTime = System.currentTimeMillis(); slotState.migratingTasks.clear(); observeLeaderMigratingStart(slot.getId()); LOGGER.info( "start migrating, slotId={}, sessionSize={}, sessions={}", slotState.slotId, sessions.size(), sessions); } final int notSyncedCount = sessions.size() - slotState.countSyncSuccess(sessions); for (String sessionIp : sessions) { MigratingTask mtask = slotState.migratingTasks.get(sessionIp); if (mtask == null) { KeyedTask<SyncSessionTask> ktask = commitSyncSessionTask(slot, slotTableEpoch, sessionIp, null, true); mtask = new MigratingTask(sessionIp, ktask); slotState.migratingTasks.put(sessionIp, mtask); LOGGER.info("migrating start,slotId={},session={}", slot.getId(), sessionIp); continue; } if (mtask.task.isFailed() && !mtask.forceSuccess) { // failed and not force Success, try to trigger emergency if (triggerEmergencyMigrating(slotState, sessions, mtask, notSyncedCount)) { LOGGER.info("[emergency]{},session={}", slotState.slotId, mtask.sessionIp); } else { KeyedTask<SyncSessionTask> ktask = commitSyncSessionTask(slot, slotTableEpoch, sessionIp, null, true); mtask.task = ktask; mtask.tryCount++; LOGGER.error( "migrating retry,slotId={},try={},session={},create={}/{}", slot.getId(), mtask.tryCount, sessionIp, mtask.createTimestamp, System.currentTimeMillis() - mtask.createTimestamp); continue; } } // force success or migrating finish. try to sync session // avoid the time of migrating is too long and block the syncing of session if (mtask.task.isOverAfter(syncSessionIntervalMs)) { if (syncSession(slotState, sessionIp, null, syncSessionIntervalMs, slotTableEpoch)) { LOGGER.info("slotId={}, sync session in migrating, session={}", slot.getId(), sessionIp); } } } } Slot 的数据的同步是一个 watchDog 的模式，获取 slotTabel 的变化，会自动同步节点上的 Slot 数据：
void syncWatch() { final int syncSessionIntervalMs = dataServerConfig.getSlotLeaderSyncSessionIntervalSecs() * 1000; final int syncLeaderIntervalMs = dataServerConfig.getSlotFollowerSyncLeaderIntervalSecs() * 1000; final long slotTableEpoch = slotTableStates.table.getEpoch(); 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); } } } 由于节点上线后，Meta 会感知并获取最新的 DataServer 节点，leader Meta 对路由表进行了刷新，因为重新分配 SlotTable 本质上是 对 DataServer 节点和 Slot 槽位之间的映射关系进行重新分配。获取到当前存活的 DataServer 节点信息，从而方便的对之进行 Slot 分配。接下来对 Server 节点的 Slot 数据进行了同步。
上面的两部分代码总体可以概括为以下两个层次：
Data 和 Session 通过心跳的机制在 Meta 上进行续约，当 Data 发生节点变更的时候，Meta 此时会重新进行分配，生成新的 SlotTable，并通过广播和心跳的方式返回所有的节点，Session 就会用这份 SlotTable 寻址新的 Data 节点。存在一个时刻，集群中存在两份 SlotTable，分裂时间最大(1s)。
Session 上会缓存 Client 的 Pub 和 Sub 数据作为基准数据，在增量发送给 Data 的同时，Slot 的 leader 节点会定时和 Session 进行数据比对，Slot 的 follower 和 leader 也会定时进行数据对比，这样做到整个集群数据能快速达到最终一致，减少异常场景的不可服务时间。
Slot 数据完整度校验的作用在于当一个 Slot 数据所有副本所在节点全部宕机，数据发生丢失时，只能通过 Session 上的缓存数据进行数据回放。但在完全回放完成时，Slot 是不可对外提供读取服务的，以此避免推空或者推错，从而保证了数据的完整性。
流程如下图：
server 启动的最后一步，是通知 Meta 将上线的 Data 节点从 blacklist 移除：
private void postStart() throws Throwable { startupRetryer.call( () -> { LOGGER.info("successful start data server, remove self from blacklist"); metaServerService.removeSelfFromMetaBlacklist(); return true; }); } 至此，Server 的启动完成，Data 节点的 Slot 数据也根据变更后的 slotTable 同步完成，Data 节点开始在集群内工作。