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 请求到各个节点， 3ff8 通知节点信息发生变化，推拉获取信息的最终效果是一致的。

当轮询信息返回数据节点有变化时，会向 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是空，就做如下操作：

通知该 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 eve
be8
nt = 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
ad8
+-------------------+

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

15a8
.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 |
|                            +-----------------+      |  +-------+----------+
|
ad8
|          |                                +--------------------------+
|                                                     |          |                                |[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() {

......

1044
}
}).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
56c
|
|                 |                                               |
|   +-------------+---------+               +---------------------+----+
|   |  NotifyOnlineHandler  |               |  NotifyFetchDatumHandler |
|   +-------------+---------+               +---------------+----------+
|                 ^ In Exist Server                         ^  In New Server
|                 |                                         |
|                 |                                         |
+-----------------------------------------------------------------------------+
|                                         |
|                                         |

1853
+-------+------------+                  +---------+-----------+
|   New Data Server  |                  |   Exist Data Server |
+--------------------+                  +---------------------+

手机如下：

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

0xFF 参考

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

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

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

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

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

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

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

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

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

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航