SOFARegistry 是蚂蚁金服开源的一个生产级、高时效、高可用的服务注册中心。本系列文章重点在于分析设计和架构，即利用多篇文章，从多个角度反推总结 DataServer 或者 SOFARegistry 的实现机制和架构思路，让大家借以学习阿里如何设计。本文为第十二篇，上文我们简述了Data节点变化之后，在dataServer中是如何变化处理的，本文我们按照数据流程继续进行，讲讲SOFARegistry如何处理本机房Data节点变化。

[从源码学设计]蚂蚁金服SOFARegistry 之 LocalDataServerChangeEvent及数据同步

目录

0x00 摘要

SOFARegistry 是蚂蚁金服开源的一个生产级、高时效、高可用的服务注册中心。

本系列文章重点在于分析设计和架构，即利用多篇文章，从多个角度反推总结 DataServer 或者 SOFARegistry 的实现机制和架构思路，让大家借以学习阿里如何设计。

本文为第十二篇，上文我们简述了Data节点变化之后，在dataServer中是如何变化处理的，本文我们按照数据流程继续进行，讲讲SOFARegistry如何处理本机房Data节点变化。

0x02 业务范畴

2.1 DataServer 数据一致性

DataServer 在 SOFARegistry 中，承担着核心的数据存储功能。数据按 dataInfoId 进行一致性 Hash 分片存储，支持多副本备份，保证数据高可用。这一层可随服务数据量的规模的增长而扩容。

如果 DataServer 宕机，MetaServer 能感知，并通知所有 DataServer 和 SessionServer，数据分片可 failover 到其他副本，同时 DataServer 集群内部会进行分片数据的迁移。

2.2 本地机房策略

Data Center 代表的是本地机房，从目前来看：

  • 数据备份仅仅在本地机房完成；
  • 每个数据中心有自己的hash；

阿里有异机房备份，应该就是Global部分，但是没有开源。

所以我们得重点剖析本地节点如何继续处理。

0x03 总体逻辑

我们先要提前剧透下总体逻辑，DataServer彼此通知是围绕着数据版本号进行，即：

  • 新加入节点 会通过 NotifyOnlineRequest 告诉其他已经在线的节点，我是新的，你可以做相应配置，以及告诉我，你有哪些版本号的数据。
  • 在线服务节点 会通过 NotifyFetchDatumRequest 告诉新节点，我这里有你需要的版本号数据，你过来取。

所以我们总结如下：在收到 meta server 的 data server change 消息之后，同一个Data Center 之中所有data server 会互相通知彼此升级版本号。

  • notifyOnline 会发送 NotifyOnlineRequest，而其他 Data Server 的 NotifyOnlineHandler 会做相应处理。
  • notifyToFetch 会发送 NotifyFetchDatumRequest，而其他 Data Server 的 notifyFetchDatumHandler 会做相应处理。

0x04 消息

4.1 LocalDataServerChangeEvent

前文提到，在DataServerChangeEventHandler中，处理DataServerChangeEvent时，若当前节点是 DataCenter 节点，则触发 LocalDataServerChangeEvent 事件。

public class LocalDataServerChangeEvent implements Event {
    private Map<String, DataNode> localDataServerMap;
    private long                  localDataCenterversion;
    private Set<String>           newJoined;
    private long                  version;
}

4.2 来源

LocalDataServerChangeEvent消息来源是DataServerChangeEventHandler。

MetaServer 会通过网络连接感知到新节点上线或者下线，所有的 DataServer 中运行着一个定时刷新连接的任务 ConnectionRefreshTask，该任务定时去轮询 MetaServer，获取数据节点的信息。需要注意的是，除了 DataServer 主动去 MetaServer 拉取节点信息外，MetaServer 也会主动发送 NodeChangeResult 请求到各个节点，通知节点信息发生变化，推拉获取信息的最终效果是一致的。

当轮询信息返回数据节点有变化时，会向 EventCenter 投递一个 DataServerChangeEvent 事件，在该事件的处理器中，如果判断出是当前机房节点信息有变化，则会投递新的事件 LocalDataServerChangeEvent，该事件的处理器 LocalDataServerChangeEventHandler 中会判断当前节点是否为新加入的节点，如果是新节点则会向其它节点发送 NotifyOnlineRequest 请求，如图所示：

在 DataServerChangeEventHandler 的 doHandle 函数中，会产生 LocalDataServerChangeEvent。

0x05 消息处理

LocalDataServerChangeEventHandler是同机房数据节点变更事件处理器，或者说是同一集群数据同步器。

LocalDataServerChangeEvent事件的处理器 LocalDataServerChangeEventHandler 中会判断当前节点是否为新加入的节点，如果是新节点则会向其它节点发送 NotifyOnlineRequest 请求。所以是针对本 Data Center中新加入的 Data Server 进行处理。

5.1 LocalDataServerChangeEventHandler

LocalDataServerChangeEventHandler 之中关键是：

private BlockingQueue<LocalDataServerChangeEvent> events = new LinkedBlockingDeque<>(); 

private class LocalClusterDataSyncer implements Runnable

讲解如下：

  • afterPropertiesSet 之后会start一个线程LocalClusterDataSyncer，用来异步处理；
  • doHandle时候，通过events来调用LocalClusterDataSyncer进行异步处理；

即在LocalDataServerChangeEventHandler内部做了一个统一延迟异步处理。当从 EventCenter 中拿到 LocalDataServerChangeEvent 之后，会往 events 放入这个 event，然后内部的LocalClusterDataSyncer 会在随后异步执行。

在LocalClusterDataSyncer内部是：

  • 如果本身是工作状态，就开始比较数据，通知相关的 Data Servers。即if local server is working, compare sync data；
  • 如果本身不是工作状态，说明自己本身就是一个新server，则通知其他server。即if local server is not working, notify others that i am newer；

LocalDataServerChangeEventHandler定义如下：

public class LocalDataServerChangeEventHandler extends
                                              AbstractEventHandler<LocalDataServerChangeEvent> {

    @Autowired
    private DataServerConfig                          dataServerConfig;

    @Autowired
    private LocalDataServerCleanHandler               localDataServerCleanHandler;

    @Autowired
    private DataServerCache                           dataServerCache;

    @Autowired
    private DataNodeExchanger                         dataNodeExchanger;

    @Autowired
    private DataNodeStatus                            dataNodeStatus;

    @Autowired
    private DatumCache                                datumCache;

    @Autowired
    private DatumLeaseManager                         datumLeaseManager;
  
    private BlockingQueue<LocalDataServerChangeEvent> events    = new LinkedBlockingDeque<>();  
}

5.1.1 投放消息

在doHandle函数中，会把最新的消息投放到 BlockingQueue 之中。

public void doHandle(LocalDataServerChangeEvent localDataServerChangeEvent) {
    isChanged.set(true);

    // Better change to Listener pattern
    localDataServerCleanHandler.reset();
    datumLeaseManager.reset();

    events.offer(localDataServerChangeEvent);
}

5.1.2 启动引擎

消费引擎在Bean启动之后，会通过afterPropertiesSet来启动。

@Override
public void afterPropertiesSet() throws Exception {
		super.afterPropertiesSet();
    start();
}

public void start() {
    Executor executor = ExecutorFactory
        .newSingleThreadExecutor(LocalDataServerChangeEventHandler.class.getSimpleName());
    executor.execute(new LocalClusterDataSyncer());
}

LocalClusterDataSyncer会执行具体业务消费消息。

0x06 消费通知消息

在引擎之中，LocalClusterDataSyncer会持续消费。

private class LocalClusterDataSyncer implements Runnable {

    @Override
    public void run() {
        while (true) {
            try {
                LocalDataServerChangeEvent event = events.take();
                //if size of events is greater than 0, not handle and continue, only handle the last one in the queue
                if (events.size() > 0) {
                    continue;
                }
                long changeVersion = event.getVersion();
                isChanged.set(false);
                if (LocalServerStatusEnum.WORKING == dataNodeStatus.getStatus()) {
                    //if local server is working, compare sync data
                    notifyToFetch(event, changeVersion);
                } else {
                    dataServerCache.checkAndUpdateStatus(changeVersion);
                    //if local server is not working, notify others that i am newer
                    notifyOnline(changeVersion);

                    dataServerCache.updateItem(event.getLocalDataServerMap(),
                        event.getLocalDataCenterversion(),
                        dataServerConfig.getLocalDataCenter());
                }
            } 
        }
    }

【重点说明】

每个data server 都会从 meta server 接收到 DataServerChangeEvent，因为是本地Data Server的消息，所以都会转换为 LocalDataServerChangeEvent。

因为是每个 data server 都会接收到，所以新上线服务器会接收到，已经在线的服务器也会接收到。这是下面讲解的重点。

6.1 新节点

在新节点中，LocalDataServerChangeEvent事件的处理器 LocalDataServerChangeEventHandler 中会判断当前节点是否为新加入的节点，如果是新节点则会向其它节点发送 NotifyOnlineRequest 请求，如图所示：

图 DataServer 节点上线时新节点的逻辑

上图展示的是新加入节点收到节点变更消息的处理逻辑，如果是线上已经运行的节点收到节点变更的消息，前面的处理流程都相同，不同之处在于 LocalDataServerChangeEventHandler 中会根据 Hash 环计算出变更节点（扩容场景下，变更节点是新节点，缩容场景下，变更节点是下线节点在 Hash 环中的后继节点）所负责的数据分片范围和其备份节点。

新加入节点 会通过 NotifyOnlineRequest 告诉其他已经在线的节点，我是新的，你可以做相应配置。

6.1.1 notifyOnline

notifyOnline 会从 DataServerNodeFactory 获取当前Local Data Center中所有的DataServerNode，然后逐一发送 NotifyOnlineRequest 通知：我上线了。

然后其他在线的Data Server 当收到通知，会开始与新节点交互。

/**
 * notify other dataservers that this server is online newly
 *
 * @param changeVersion
 */
private void notifyOnline(long changeVersion) {
    Map<String, DataServerNode> dataServerNodeMap = DataServerNodeFactory
        .getDataServerNodes(dataServerConfig.getLocalDataCenter());
    for (Entry<String, DataServerNode> serverEntry : dataServerNodeMap.entrySet()) {
        while (true) {
            String ip = serverEntry.getKey();
            DataServerNode dataServerNode = DataServerNodeFactory.getDataServerNode(
                dataServerConfig.getLocalDataCenter(), ip);
            try {
                final Connection connection = dataServerNode.getConnection();
                CommonResponse response = (CommonResponse) dataNodeExchanger.request(
                    new Request() {

                        @Override
                        public Object getRequestBody() {
                            return new NotifyOnlineRequest(DataServerConfig.IP,
                                changeVersion);
                        }

                        @Override
                        public URL getRequestUrl() {
                            return new URL(connection.getRemoteIP(), connection
                                .getRemotePort());
                        }
                    }).getResult();
            } 
        }
    }
}

6.2 在线服务节点

当前在线服务节点遍历自身内存中的数据项，过滤出属于变更节点的分片范围的数据项，然后向变更节点和其备份节点发送 NotifyFetchDatumRequest 请求， 变更节点和其备份节点收到该请求后，其处理器会向发送者同步数据（NotifyFetchDatumHandler.fetchDatum），如图所示。

注意，本图与上图左右的JVM放置位置相反。

图 DataServer 节点变更时已存节点的逻辑

就是说，在线服务节点 会通过 NotifyFetchDatumRequest 告诉新节点，我这里有你需要的数据，你过来取。

下面是几个重要函数的说明：

6.2.1 notifyToFetch

notify onlined newly dataservers to fetch datum，就是通知新节点，你主动过来拉取，同时也依据请求消息来更新自身。

notifyToFetch的具体功能是：

  • 首先从 event 获取新Server，这一份数据被设置成三种格式，分别是 Map 格式dataServerMapIn 和 list 格式 dataServerNodeList，ConcurrentHashMap格式的dataServerMap；
  • 用新Server生成一个consistentHash；
  • 使用 toBeSyncMap = getToBeSyncMap(consistentHash); 获取需要同步的map；getToBeSyncMap的作用是哪些ip需要同步哪些东西；get map of datum to be synced。
    • 遍历 toBeSyncMap，对于其中每一个需要同步的 toBeSyncEntry，获取其IP和dataInfoMap，dataInfoMap 是Map<String, Map<String, BackupTriad>> 类型；
      • 遍历 dataInfoMap 中所有的 Entry<String, Map<String, BackupTriad>> dataCenterEntry，该entry 的key 是dataCenter；
        • 遍历 dataTriadEntry 中所有的Entry<String, BackupTriad> dataTriadEntry，其key是 dataInfoId；
          • 利用 dataInfoId 从 datumCache 中获取 Datum；
          • 获取 Datum 版本号 versionMap.put(dataInfoId, datum.getVersion());
      • 针对这个 dataCenter 构建一个统一的大版本号map： allVersionMap，allVersionMap.put(dataCenter, versionMap);
      • 如果allVersionMap是空，就做如下操作：
        • 从dataServerCache中移除对应IP；
        • 通知该 ip对应的 data server，你需要同步这些：doNotify(ip, allVersionMap, changeVersion)；即告诉这个ip，你需要同步这个 dataCenter 中的这些 dataInfoId，以及其版本号；
        • 从dataServerCache中移除ip；
  • 如果 ConcurrentHashMap格式的dataServerMap 非空，遍历其key，这是一个targetIp，从dataServerCache中移除targetIp；
  • dataServerCache 根据 dataServerMapIn 更新server list；

6.2.2 getToBeSyncMap

getToBeSyncMap 的逻辑是找出需要通知的IP列表，以及每个ip需要同步哪些dataInfoId，具体如下：

  • 函数参数是根据新servers 新计算出来的 consistentHashs
  • 根据 dataServerConfig 的旧配置计算一个旧的hash，consistentHashOld；
  • 对于 datumCache 的每一个Datum，计算新的triad；具体如下：
  • 获取 datumCache 所有数据，构建一个 allMap，遍历 allMap 中所有 dataCenterEntry：
    • 对于该数据中心，遍历该data center所有的datumMap：
    • 以 dataInfoId 遍历这个 datumMap：
      • 用新 consistentHash 计算出 新的 backupNodes；
      • 用旧 consistentHashOld 得到旧的 backupTriad；
      • 从 backupTriad 获取newJoinedNodes，即从新的 backupNodes 移除 backupTriad 和 NotWorking；
      • 遍历 newJoinedNodes，对于每个node，构建 toBeSyncMap = Map<node ip, Map<dataCenter, Map<dataInfoId, BackupTriad>>>
  • 返回 toBeSyncMap；这个就是哪些ip需要同步哪些东西；
private Map<String/*ip*/, Map<String/*datacenter*/, Map<String/*datainfoId*/, BackupTriad>>> getToBeSyncMap(ConsistentHash<DataNode> consistentHash) {

    Map<String, Map<String, Map<String, BackupTriad>>> toBeSyncMap = new HashMap<>();
    Map<String, List<DataNode>> triadCache = new HashMap<>();

    ConsistentHash<DataNode> consistentHashOld = dataServerCache
        .calculateOldConsistentHash(dataServerConfig.getLocalDataCenter());
}

6.2.3 getNewJoined

getNewJoined就是找出那些不在已经存储的Triad 之中，或者在其中但是不是working状态的。

public List<DataNode> getNewJoined(List<DataNode> newTriad, Set<String> notWorking) {
    List<DataNode> list = new ArrayList<>();
    for (DataNode node : newTriad) {
        String ip = node.getIp();
        if (!ipSetOfNode.contains(ip) || notWorking.contains(ip)) {
            list.add(node);
        }
    }
    return list;
}

6.2.4 BackupTriad

BackupTriad 的作用是：针对 dataInfoId，对应的备份DataNode列表。

public class BackupTriad {
    /** dataInfoId */
    private String         dataInfoId;

    /**
     * calculate current dataServer list Consistent hash to get dataInfoId belong node and backup node list
     * @see  ConsistentHash#ConsistentHash(int, java.util.Collection)
     * @see  com.alipay.sofa.registry.consistency.hash.ConsistentHash#getNUniqueNodesFor(java.lang.Object, int)
     */
    private List<DataNode> triad;

    private Set<String>    ipSetOfNode = new HashSet<>();

    /**
     * constructor
     * @param dataInfoId
     * @param triad
     */
    public BackupTriad(String dataInfoId, List<DataNode> triad) {
        this.dataInfoId = dataInfoId;
        this.triad = triad;
        for (DataNode node : triad) {
            ipSetOfNode.add(node.getIp());
        }
    }
}

运行时如下：

backupTriad = {BackupTriad@1400} "BackupTriad{dataInfoId='TestDataInfoId', ipSetOfNode=[192.168.0.2, 192.168.0.1, 192.168.0.3]}"
 dataInfoId = "TestDataInfoId"
 triad = {ArrayList@1399}  size = 3
  0 = {DataNode@1409} "DataNode{ip=192.168.0.1}"
  1 = {DataNode@1410} "DataNode{ip=192.168.0.2}"
  2 = {DataNode@1411} "DataNode{ip=192.168.0.3}"
 ipSetOfNode = {HashSet@1403}  size = 3
  0 = "192.168.0.2"
  1 = "192.168.0.1"
  2 = "192.168.0.3"

0x07 changeVersion 从哪里来

在上述代码中，会从LocalDataServerChangeEvent获取一个version，从而利用这个版本做后续处理，同时也会给dataServerCache设置版本号。

LocalDataServerChangeEvent event = events.take();
long changeVersion = event.getVersion();
if (LocalServerStatusEnum.WORKING == dataNodeStatus.getStatus()) {
    //if local server is working, compare sync data
    notifyToFetch(event, changeVersion);
} else {
    dataServerCache.checkAndUpdateStatus(changeVersion);
    //if local server is not working, notify others that i am newer
    notifyOnline(changeVersion);
}

现在我们就好奇，当Data Server有变更时候，这个版本是从哪里来的。让我们追根溯源。这是从后往前倒推的过程。

7.1 版本号和变化

7.1.1 DataServerCache

因为提到了dataServerCache设置版本号，所以我们要回溯到DataServerCache。可以看到，DataServerCache之中有两个相关变量：curVersion 和 DataServerChangeItem。

这就是从newDataServerChangeItem获取了对应data center的版本号，设置在DataServerCache。

具体DataServerCache中相关定义如下：

public class DataServerCache {
  
    /** new input dataServer list and version */
    private volatile DataServerChangeItem                 newDataServerChangeItem = new DataServerChangeItem();
  
    private AtomicLong                                    curVersion              = new AtomicLong(-1L);

    public Long getDataCenterNewVersion(String dataCenter) {
        synchronized (DataServerCache.class) {
            Map<String, Long> versionMap = newDataServerChangeItem.getVersionMap();
            if (versionMap.containsKey(dataCenter)) {
                return versionMap.get(dataCenter);
            } else {
                return null;
            }
        }
    }  
}

7.1.2 设置和使用

在 DataServerCache中只有addStatus控制curVersion的赋值，而对外的接口中，只有 synced 和 addNotWorkingServer 调用 addStatus。

而 newDataServerChangeItem 是在compareAndSet这里设置。

public Map<String, Set<String>> compareAndSet(DataServerChangeItem newItem, FromType fromType) {
            if (!changedMap.isEmpty()) {
                newDataServerChangeItem = newItem;
            }
}

逻辑如下：

                        +-----------------------------+
                        |[DataServerCache]            |
                        |                             |
compareAndSet +-------------> DataServerChangeItem    |
                        |                             |
                        |     curVersion              |
                        |     ^        ^              |
                        |     |        |              |
                        +-----------------------------+
                              |        |
synced +----------------------+        |
                                       |
addNotWorkingServer+-------------------+

7.1.3 两个设计点

现在涉及到 DataServerCache 两个设计点：

  • curVersion 是用来做什么的；
  • newDataServerChangeItem是用来做什么的；

现在推论，每一个数据中心 Data Center 有一个版本号用做其内部所有状态控制。其实，在DataServerChangeItem 的定义中的 versionMap 也能看出来，是根据版本号控制的。

DataServerChangeItem 定义如下：

public class DataServerChangeItem {

    /** datacenter -> Map<ip, DataNode> */
    private Map<String, Map<String, DataNode>> serverMap;

    /** datacenter -> version */
    private Map<String, Long>                  versionMap;
}

从而知道：

  • curVersion 就是Data Center最新的版本号；
  • newDataServerChangeItem就是最新版本号对应的变化数据；

现在问题变成，

  • DataServerChangeItem是从哪里来的。
  • curVersion 是从哪里来的。

我们通过研读源码可以知道，是从Meta Server获取，下面就跟踪下这个过程。

7.2 Data Server

7.2.1 主动获取变化

我们需要复习下这个流程。

Meta Server 会广播通知 所有data server 现在有 data server 更新，也可能是 DataServer主动定期看看MetaServer 是否有更新。

但是具体更新的内容，还是 data server 主动发送 GetNodesRequest 获取。

这里以主动更新为例，可以看到，DataServer 会通过 metaServerService.getDateServers 从 meta server 获取到DataServerChangeItem，从而构建 DataServerChangeEvent。

public class ConnectionRefreshTask extends AbstractTask {

    @Autowired
    private IMetaServerService metaServerService;

    @Autowired
    private EventCenter        eventCenter;

    @Override
    public void handle() {
        DataServerChangeItem dataServerChangeItem = metaServerService.getDateServers();
        if (dataServerChangeItem != null) {
            eventCenter
                .post(new DataServerChangeEvent(dataServerChangeItem, FromType.CONNECT_TASK));
        }
    }
}

在 DefaultMetaServiceImpl 中可以看到，DataServerChangeItem是从 Meta Server获取的NodeChangeResult 提取出来。

public class DefaultMetaServiceImpl implements IMetaServerService {
    @Override
    public DataServerChangeItem getDateServers() {
        Map<String, Connection> connectionMap = metaServerConnectionFactory
            .getConnections(dataServerConfig.getLocalDataCenter());
        String leader = getLeader().getIp();
        if (connectionMap.containsKey(leader)) {
            Connection connection = connectionMap.get(leader);
            if (connection.isFine()) {
                try {
                    GetNodesRequest request = new GetNodesRequest(NodeType.DATA);
                    Object obj = metaNodeExchanger.request(new Request() {
                        @Override
                        public Object getRequestBody() {
                            return request;
                        }

                        @Override
                        public URL getRequestUrl() {
                            return new URL(connection.getRemoteIP(), connection.getRemotePort());
                        }
                    }).getResult();
                    if (obj instanceof NodeChangeResult) {
                        NodeChangeResult<DataNode> result = (NodeChangeResult<DataNode>) obj;
                        Map<String, Long> versionMap = result.getDataCenterListVersions();
                        versionMap.put(result.getLocalDataCenter(), result.getVersion());
                        return new DataServerChangeItem(result.getNodes(), versionMap);
                    }
                } 
            }
        }
        String newip = refreshLeader().getIp();
        return null;
    }  
}

逻辑如下：

+  Data Server
|
|
|  +------------------+
|  | NodeChangeResult |
|  +-------+----------+
|          |                                +--------------------------+
|          |                                |[DataServerCache]         |
|          |                                |                          |
|          +---------------->compareAndSet------> DataServerChangeItem |
|     DataServerChangeItem                  |                          |
|                                           |     curVersion           |
|                                           |     ^        ^           |
|                                           |     |        |           |
|                                           +--------------------------+
|                                                 |        |
|                             synced +-------------        |
|                                                          |
|                             addNotWorkingServer----------+
|
|
+

7.3 Meta server

7.3.1 设置版本号

让我们来到 meta server之中。可以看到，之前 在DataStoreService put，remove等各个函数中，当data server有变化时候，会调用 dataNodeRepository 通过时间戳设置版本号。

dataNodeRepository.setVersion(currentTimeMillis);

7.3.2 提取版本号

当 meta server 接收到 GetNodesRequest 之后，会生成 NodeChangeResult。

DataStoreService 会调用 dataNodeRepository 获取版本号，从而在 NodeChangeResult之中设置。

public class DataStoreService implements StoreService<DataNode> {
    @Override
    public NodeChangeResult getNodeChangeResult() {

        NodeChangeResult nodeChangeResult = new NodeChangeResult(NodeType.DATA);

        try {
            String localDataCenter = nodeConfig.getLocalDataCenter();
            Map<String/*dataCenter*/, NodeRepository> dataNodeRepositoryMap = dataRepositoryService
                    .getNodeRepositories();

            ConcurrentHashMap<String/*dataCenter*/, Map<String/*ipAddress*/, DataNode>> pushNodes = new ConcurrentHashMap<>();
            Map<String/*dataCenter*/, Long> versionMap = new ConcurrentHashMap<>();

            dataNodeRepositoryMap.forEach((dataCenter, dataNodeRepository) -> {
               //在这里会获取版本号
                if (localDataCenter.equalsIgnoreCase(dataCenter)) {                
                    nodeChangeResult.setVersion(dataNodeRepository.getVersion());
                }
            
                versionMap.put(dataCenter, dataNodeRepository.getVersion());

                Map<String, RenewDecorate<DataNode>> dataMap = dataNodeRepository.getNodeMap();
                Map<String, DataNode> newMap = new ConcurrentHashMap<>();
                dataMap.forEach((ip, dataNode) -> newMap.put(ip, dataNode.getRenewal()));
                pushNodes.put(dataCenter, newMap);
            });

            nodeChangeResult.setNodes(pushNodes);
            nodeChangeResult.setDataCenterListVersions(versionMap);
            nodeChangeResult.setLocalDataCenter(localDataCenter);
        } 
				//返回
        return nodeChangeResult;
    }  
}

具体如下：

                                                   Meta Server  +  Data Server
                                                                |
                                                                |
             getNodeChangeResult       +-----------------+      |  +------------------+
          +------------------------->  | NodeChangeResult| +------>+ NodeChangeResult |
          |                            +-----------------+      |  +-------+----------+
          |                                                     |          |                                +--------------------------+
          |                                                     |          |                                |[DataServerCache]         |
 +--------+--------+                                            |          |                                |                          |
 |DataStoreService | +-------------------+                      |          +---------------->compareAndSet------> DataServerChangeItem |
 +-----------------+    getVersion       |                      |     DataServerChangeItem                  |                          |
                                         |                      |                                           |     curVersion           |
                                         |                      |                                           |     ^        ^           |
                                         |                      |                                           |     |        |           |
                                         v                      |                                           +--------------------------+
+----------------------+               +-+-----------------+    |                                                 |        |
| DataRepositoryService+-------------> |dataNodeRepository |    |                             synced +-------------        |
+----------------------+               +-------------------+    |                                                          |
                setVersion(currentTimeMillis)                   |                             addNotWorkingServer----------+
                                                                |
                                                                |
                                                                +

手机上如图：

7.4 Data Server

7.4.1 获得变化

我们又回到 Data Server。

当Data Server接收到NodeChangeResult之后，会提取出DataServerChangeItem。

public class DefaultMetaServiceImpl implements IMetaServerService {
  
    @Override
    public DataServerChangeItem getDateServers() {
   
                    ......
      
                    GetNodesRequest request = new GetNodesRequest(NodeType.DATA);
                    Object obj = metaNodeExchanger.request(new Request() {
                      
                        ......
                      
                        }
                    }).getResult();
      
                    if (obj instanceof NodeChangeResult) {
                        NodeChangeResult<DataNode> result = (NodeChangeResult<DataNode>) obj;
                        Map<String, Long> versionMap = result.getDataCenterListVersions();
                        versionMap.put(result.getLocalDataCenter(), result.getVersion());
                        return new DataServerChangeItem(result.getNodes(), versionMap);
                    }
                } 
            }
        }
    }  
}

然后会回到前面 “主动获取变化” 小节，发送DataServerChangeEvent，进而转化为LocalDataServerChangeEvent，就和我们的代码联系起来。

0x08 Data Server后续处理

关于 DataServerCache . curVersion 和 newDataServerChangeItem 如何进一步处理，我们需要再研究。

8.1 newDataServerChangeItem

DataServerChangeEventHandler 的 doHandle 函数中有使用：

for (Entry<String, Set<String>> changeEntry : changedMap.entrySet()) {
    String dataCenter = changeEntry.getKey();
    Set<String> ips = changeEntry.getValue();
    Long newVersion = dataServerCache.getDataCenterNewVersion(dataCenter);
}

调用的是dataServerCache的函数，可以看到是取出newDataServerChangeItem的版本号。

public Long getDataCenterNewVersion(String dataCenter) {
    synchronized (DataServerCache.class) {
        Map<String, Long> versionMap = newDataServerChangeItem.getVersionMap();
        if (versionMap.containsKey(dataCenter)) {
            return versionMap.get(dataCenter);
        } else {
            return null;
        }
    }
}

构建LocalDataServerChangeEvent时候，则把newDataServerChangeItem的版本作为本地版本号localDataCenterversion。

public LocalDataServerChangeEvent(Map<String, DataNode> localDataServerMap,
                                  Set<String> newJoined, long version,
                                  long localDataCenterversion) {
    this.localDataServerMap = localDataServerMap;
    this.newJoined = newJoined;
    this.version = version;
    this.localDataCenterversion = localDataCenterversion;
}

dataServerCache会据此做相关更新。

dataServerCache.updateItem(dataServerMapIn, event.getLocalDataCenterversion(),
    dataServerConfig.getLocalDataCenter());

8.2 curVersion

关于curVersion，则来到了 notifyToFetch 和 notifyOnline 后续如何处理。

8.2.1 发送版本号

前面我们只是讲解了如何发送版本号，即：

  • 在线服务节点 会通过 NotifyFetchDatumRequest 告诉新节点，我这里有你需要的数据，你过来取。
  • 新加入节点 会通过 NotifyOnlineRequest 告诉其他已经在线的节点，我是新的，你可以做相应配置。

所以我们总结可以，在收到 meta server 的 data server change 消息之后，同一个Data Center 之中所有data server 会互相通知彼此升级版本号。

  • notifyOnline 会发送 NotifyOnlineRequest，而其他 Data Server 的 NotifyOnlineHandler 会做相应处理。

  • notifyToFetch 会发送 NotifyFetchDatumRequest，而其他 Data Server 的 notifyFetchDatumHandler 会做相应处理。

8.2.2 接收版本号

下面我们要看看接收版本号之后，DataServer的新节点与在线节点分别做了什么。

  • notifyFetchDatumHandler—-新节点处理

这是一个数据拉取请求，当该 Handler 被触发时，通知当前 DataServer 节点进行版本号对比，若请求中数据的版本号高于当前节点缓存中的版本号，则会进行数据同步操作，保证数据是最新的。

  • notifyOnlineHandler—-在线节点处理

这是一个 DataServer 上线通知请求 Handler，当其他节点上线时，会触发该 Handler，从而当前节点在缓存中存储新增的节点信息。用于管理节点状态，究竟是 INITIAL 还是 WORKING 。

于是可以看到，在NotifyOnlineHandler和NotifyFetchDatumHandler之中，都会根据本地dataServerCache中存储的curVersion做判断是否需要继续处理。

public class NotifyOnlineHandler extends AbstractServerHandler<NotifyOnlineRequest> {

    @Autowired
    private DataServerCache dataServerCache;

    @Override
    public Object doHandle(Channel channel, NotifyOnlineRequest request) {
        long version = request.getVersion();
        if (version >= dataServerCache.getCurVersion()) {
            dataServerCache.addNotWorkingServer(version, request.getIp());
        }
        return CommonResponse.buildSuccessResponse();
    }
}

以及 NotifyFetchDatumHandler 之中会调用sycned。

public class NotifyFetchDatumHandler extends AbstractServerHandler<NotifyFetchDatumRequest> {

    private static final Logger         LOGGER = LoggerFactory
                                                   .getLogger(NotifyFetchDatumHandler.class);

    @Autowired
    private DataServerCache             dataServerCache;

    @Autowired
    private DataServerConnectionFactory dataServerConnectionFactory;

    @Autowired
    private DataChangeEventCenter       dataChangeEventCenter;

    @Autowired
    private Exchange                    boltExchange;

    @Autowired
    private DataServerConfig            dataServerConfig;

    @Autowired
    private DatumCache                  datumCache;

    @Autowired
    private LocalDataServerCleanHandler localDataServerCleanHandler;

    @Override
    public Object doHandle(Channel channel, NotifyFetchDatumRequest request) {
        ParaCheckUtil.checkNotBlank(request.getIp(), "ip");

        //receive other data NotifyFetchDatumRequest,must delay clean datum task until fetch all datum
        localDataServerCleanHandler.reset();

        Map<String, Map<String, Long>> versionMap = request.getDataVersionMap();
        long version = request.getChangeVersion();
        String ip = request.getIp();
        if (version >= dataServerCache.getCurVersion()) {
            if (versionMap.isEmpty()) {
                dataServerCache.synced(version, ip);
            } else {
                ExecutorFactory.getCommonExecutor().execute(() -> {
                    for (Entry<String, Map<String, Long>> dataCenterEntry : versionMap.entrySet()) {
                        String dataCenter = dataCenterEntry.getKey();
                        Map<String, Long> map = dataCenterEntry.getValue();
                        for (Entry<String, Long> dataInfoEntry : map.entrySet()) {
                            String dataInfoId = dataInfoEntry.getKey();
                            Datum datum = datumCache.get(dataCenter, dataInfoId);
                            if (datum != null) {
                                long inVersion = dataInfoEntry.getValue();
                                long currentVersion = datum.getVersion();
                                if (currentVersion > inVersion) {
                                    continue;
                                } else if (datum.getVersion() == dataInfoEntry.getValue()) {
                                    //if version same,maybe remove publisher all by LocalDataServerCleanHandler,so must fetch from other node
                                    if (!datum.getPubMap().isEmpty()) {
                                        continue;
                                    }
                                }
                            }
                            fetchDatum(ip, dataCenter, dataInfoId);
                        }
                    }
                    dataServerCache.synced(version, ip);
                });
            }
        } 
        return CommonResponse.buildSuccessResponse();
    }
}

于是，目前如下：

                                                                +
                                                                |
                                                   Meta Server  |  Data Server
                                                                |
                                                                |
             getNodeChangeResult       +-----------------+      |  +------------------+
          +------------------------->  | NodeChangeResult| +------>+ NodeChangeResult |
          |                            +-----------------+      |  +-------+----------+
          |                                                     |          |                                +--------------------------+
          |                                                     |          |                                |[DataServerCache]         |
 +--------+--------+                                            |          |                                |                          |
 |DataStoreService | +-------------------+                      |          +---------------->compareAndSet+-----> DataServerChangeItem |
 +-----------------+    getVersion       |                      |     DataServerChangeItem                  |                          |
                                         |                      |                                           |     curVersion           |
                                         |                      |                                           |     ^        ^           |
                                         |                      |                                           |     |        |           |
                                         v                      |                                           +-----------------+---+----+
+----------------------+               +-+-----------------+    |                                                 |        |  ^   ^
| DataRepositoryService+-------------> |dataNodeRepository |    |     synced +------------------------------------+        |  |   | getCurVersion
+----------------------+               +-------------------+    |                                                          |  |   |
                setVersion(currentTimeMillis)                   |     addNotWorkingSer^er+---------------------------------+  |   |
                                                                |                 +-------------------------------------------+   |
                                                                |                 | getCurVersion                                 |
                                                                |                 |                                               |
                                                                |   +-------------+---------+               +---------------------+----+
                                                                |   |  NotifyOnlineHandler  |               |  NotifyFetchDatumHandler |
                                                                |   +-------------+---------+               +---------------+----------+
                                                                |                 ^ In Exist Server                         ^  In New Server
                                                                |                 |                                         |
                                                                |                 |                                         |
                                                                +-----------------------------------------------------------------------------+
                                                                                  |                                         |
                                                                                  |                                         |
                                                                          +-------+------------+                  +---------+-----------+
                                                                          |   New Data Server  |                  |   Exist Data Server |
                                                                          +--------------------+                  +---------------------+

手机如下：

至此，版本号流程就完全梳理完毕。

0xFF 参考

蚂蚁金服服务注册中心如何实现 DataServer 平滑扩缩容

蚂蚁金服服务注册中心 SOFARegistry 解析 | 服务发现优化之路

服务注册中心 Session 存储策略 | SOFARegistry 解析

海量数据下的注册中心 – SOFARegistry 架构介绍

服务注册中心数据分片和同步方案详解 | SOFARegistry 解析

蚂蚁金服开源通信框架SOFABolt解析之连接管理剖析

蚂蚁金服开源通信框架SOFABolt解析之超时控制机制及心跳机制

蚂蚁金服开源通信框架 SOFABolt 协议框架解析

蚂蚁金服服务注册中心数据一致性方案分析 | SOFARegistry 解析

版权声明：本文为rossiXYZ原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/rossiXYZ/p/14198695.html