[源码阅读] 阿里SOFA服务注册中心MetaServer(2)

2021-06-30 02:33:38

[源码阅读] 阿里SOFA服务注册中心MetaServer(2)

[源码阅读] 阿里SOFA服务注册中心MetaServer(2)

0x00 摘要

SOFARegistry 是蚂蚁金服开源的一个生产级、高时效、高可用的服务注册中心。本系列将带领大家一起分析其MetaServer的实现机制，本文为第二篇，介绍MetaServer基本功能，比如注册，存储，通知，续约等。

0x01 MetaServer 注册

1.1 Leader入口

MetaServer 的业务启动是从setLeaderProcessListener开始。

前面提到，MetaServer 集群内部基于 Raft 协议选举和复制，只要不超过 1⁄2 节点宕机，就可以对外服务。

Raft 协议由三个部分组成：

*选举（Leader Election）
日志复制（Log Replication）
安全性（Safety）

如果使用JRaft, 需要实现其状态机，而在MetaServer之中， jraft FSM 的具体实现是 ServiceStateMachine类（后文中会有大量 Raft 相关内容）。

在 raft 选出了MetaServer leader之后，ServiceStateMachine会调用 setLeaderProcessListener，其中又调用到了registerCurrentNode，这样就在MetaServer中注册了当前Meta节点。

raftServer.setLeaderProcessListener(new LeaderProcessListener() {

    @Override

    public void startProcess() {

        executorManager.startScheduler();

        PeerId leader = new PeerId(NetUtil.getLocalAddress().getHostAddress(),

            metaServerConfig.getRaftServerPort());

        registerCurrentNode();  //

        raftServer.sendNotify(leader, "leader");

    }

其运行堆栈如下：

register:51, MetaServerRegistry (com.alipay.sofa.registry.server.meta.registry)

registerCurrentNode:203, RaftExchanger (com.alipay.sofa.registry.server.meta.remoting)

access$200:56, RaftExchanger (com.alipay.sofa.registry.server.meta.remoting)

startProcess:105, RaftExchanger$1 (com.alipay.sofa.registry.server.meta.remoting)

lambda$onLeaderStart$2:234, ServiceStateMachine (com.alipay.sofa.registry.jraft.bootstrap)

run:-1, 318295790 (com.alipay.sofa.registry.jraft.bootstrap.ServiceStateMachine$$Lambda$191)

call:511, Executors$RunnableAdapter (java.util.concurrent)

run$$$capture:266, FutureTask (java.util.concurrent)

run:-1, FutureTask (java.util.concurrent)

1.2 注册

注册行为由 Registry 完成。注册接口实现为：

public interface Registry<T extends Node> {

    NodeChangeResult setNodes(List<T> nodes);

    NodeChangeResult register(T node);

    void cancel(String connectId, NodeType nodeType);

    void evict();

    void renew(T node, int duration);

    void getOtherDataCenterNodeAndUpdate(NodeType nodeType);

    DataCenterNodes getDataCenterNodes(NodeType nodeType);

    NodeChangeResult getAllNodes(NodeType nodeType);

    void pushNodeListChange(NodeType nodeType);

}

具体实现举例是：

public class MetaServerRegistry implements Registry<Node> {

    @Override

    public NodeChangeResult register(Node node) {

        StoreService storeService = ServiceFactory.getStoreService(node.getNodeType());

        return storeService.addNode(node);

    }

}

Registry 根据不同的节点类型，获取对应的StoreService完成添加节点服务，这里节点类型就是"META"。

node = {MetaNode@6342} "MetaNode{ip=192.168.1.2}"

 nodeType = {Node$NodeType@6369} "META"

 nodeUrl = {URL@6370} "URL{address='192.168.1.2:0'}"

 dataCenter = "DefaultDataCenter"

 name = "192.168.1.2"

 regionId = null

 nodeStatus = {Node$NodeStatus@6373} "INIT"

storeService 就是从上文 ServiceFactory 介绍的 storeServiceMap 中提取。

比如 MetaNode，其对应 storeService 实现为 MetaStoreService ，所以就会用 MetaStoreService 完成存储。

storeServiceMap = {HashMap@6394}  size = 3

 {Node$NodeType@6454} "SESSION" -> {SessionStoreService@6455}

 {Node$NodeType@6369} "META" -> {MetaStoreService@6456}

 {Node$NodeType@6457} "DATA" -> {DataStoreService@6458}

1.3 存储服务

Node 然后由 StoreService 存储到 Repository 中，具体举例 MetaStoreService 实现为：

public class MetaStoreService implements StoreService<MetaNode> {

    @RaftReference(uniqueId = "metaServer")

    private RepositoryService<String, RenewDecorate<MetaNode>> metaRepositoryService;

    @Override

    public NodeChangeResult addNode(MetaNode metaNode) {

        NodeChangeResult nodeChangeResult;

        String ipAddress = metaNode.getNodeUrl().getIpAddress();

        write.lock();

        try {

            //存放到repository（自动通过jraft同步给集群）

            metaRepositoryService.put(ipAddress, new RenewDecorate(metaNode,

                RenewDecorate.DEFAULT_DURATION_SECS));

            //触发通知(需要通知data/session)

            nodeChangeResult = getNodeChangeResult();

            firePushDataListTask(nodeChangeResult, "addMetaNode");

            firePushSessionListTask(nodeChangeResult, "addMetaNode");

        } finally {

            write.unlock();

        }

        return nodeChangeResult;

    }

}

预先把存储流程总结如下：

                                                                                      +------------------------------+

                                                     +-----------------------+        |  Map(String, NodeRepository) |

                                                +--->+ metaRepositoryService +------->+  registry                    |

                                                |    +-----------------------+        +------------------------------+

                                                |

                                                |

                                                |

Register +-------------------+  addNode   +-----+-----------+

+------> | MetaServerRegistry| +--------> | MetaStoreService|

         +-------------------+            +-----+-----------+

                                                |

                                                |

                                                |            +-------------------+     +--------------+       +----------------------+

                                                +----------> |TaskListenerManager+---> |TaskDispatcher| +---> |DataNodeChangePushTask|

                                               sendTaskEvent +-------------------+     +--------------+       +----------------------+

手机上参见：

1.4 Repository服务

Repository算是一个比较经典的概念了，封装数据查询和存储逻辑。定义（来自Martin Fowler的《企业应用架构模式》）：Mediates between the domain and data mapping layers using a collection-like interface for accessing domain objects。

Repository是一个独立的层，介于领域层与数据映射层（数据访问层）之间。它的存在让领域层感觉不到数据访问层的存在，即提供一个类似集合的接口提供给领域层进行领域对象的访问。

Repository可以被认为是仓库管理员，领域层需要什么东西只需告诉仓库管理员，由仓库管理员把东西拿给它，并不需要知道东西实际放在哪。

在这里，Node并没有持久化，而是存储在内存中。于是就要 重点注意，虽然Repository存储一个Node是很简单的，但是在分布式状态下，如何保证 "在一个Repository中新Node的存储状态" 可以拓展到集群其他节点中？而且能保证数据一致性呢？

1.4.1 MetaRepositoryService

MetaRepositoryService 内部的存储是在集群内部由Raft协议来保证数据一致性的。

后端 Repository 可以看作 SOFAJRaft 的状态机，任何对 Map 的操作都会在集群内部，交由 Raft 协议进行同步，从而达到集群内部的一致。从代码中可以看到MetaRepositoryService加上了 RaftService 注解，这就是Raft的部分实现。

为了更好的说明，我们这里提前介绍下 Raft 的两个注解：RaftReference 和 RaftService。

这两个注解可以认为是封装好Raft的，呈现给Registry的接口。RaftReference 对应了客户端代理，RaftService对应着服务端的实现。为什么要这么做？因为需要维护数据一致性，所以必须把单纯的本地调用转换为异步网络调用，这样才能用raft协议保证数据一致性。

RepositoryService 的具体实现类都加了 @RaftService 注解，这样就说明自己是一个服务端；
凡是引用 RepositoryService 的地方，都加了 @RaftReference，这样调用 RepositoryService 的函数就相当于客户端调用到服务端；
凡是加了 @RaftReference 注解的属性，都会被动态代理类替换，其代理实现见 ProxyHandler 类，即将方法调用，封装为 ProcessRequest，通过 RaftClient 发送给 RaftServer。

回到MetaRepositoryService代码，具体如下：

@RaftService(uniqueId = "metaServer")

public class MetaRepositoryService extends AbstractSnapshotProcess

                          implements RepositoryService<String, RenewDecorate<MetaNode>> {

    @Autowired

    private NodeConfig                                 nodeConfig;

    /**

     * meta node store and version

     */

    private Map<String/*dataCenter*/, NodeRepository> registry          = new ConcurrentHashMap<>();

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

}

1.4.2 put操作

当有新节点时候，MetaRepositoryService 会进行 put 操作。在具体put函数调用中，实际上是将方法调用，封装为 ProcessRequest，通过 RaftClient 发送给 RaftServer。这样就由 Raft 协议保证了数据一致性。

@Override

public RenewDecorate<MetaNode> put(String ipAddress, RenewDecorate<MetaNode> metaNode,

                                   Long currentTimeMillis) {

    try {

        String dataCenter = metaNode.getRenewal().getDataCenter();

        NodeRepository<MetaNode> metaNodeRepository = registry.get(dataCenter);

        if (metaNodeRepository == null) {

            NodeRepository<MetaNode> nodeRepository = new NodeRepository<>(dataCenter,

                new ConcurrentHashMap<>(), currentTimeMillis);

            // put操作实际上是调用到了服务端

            metaNodeRepository = registry.put(dataCenter, nodeRepository);

            if (metaNodeRepository == null) {

                metaNodeRepository = nodeRepository;

            }

        }

        metaNodeRepository.setVersion(currentTimeMillis);

        Map<String/*ipAddress*/, RenewDecorate<MetaNode>> metaNodes = metaNodeRepository

            .getNodeMap();

        RenewDecorate oldRenewDecorate = metaNodes.get(ipAddress);

        metaNodes.put(ipAddress, metaNode);

    }

    return metaNode;

}

1.4.3 节点数据存储

节点数据的存储，其本质上是存储在内存的哈希表中，其存储结构为：

// RepositoryService 底层存储

Map<String/*dataCenter*/, NodeRepository> registry;

// NodeRepository 底层存储

Map<String/*ipAddress*/, RenewDecorate<T>> nodeMap;

将RenewDecorate存储到该 Map 中，整个节点注册的流程就完成了，至于如何和 Raft 协议进行结合和数据同步，后续会介绍。

节点移除的逻辑类似，将节点信息从该 Map 中删除，也会存储一个变更事件到队列。

最后结果如下：

this = {MetaRepositoryService@6905}

 registry = {ConcurrentHashMap@6907}  size = 1

  "DefaultDataCenter" -> {NodeRepository@7251}

 snapShotFileNames = {HashSet@6904}  size = 1

1.5 jraft实现数据一致性

成员列表数据存储在 Repository 中，Repository 被一致性协议层进行包装，作为 SOFAJRaft 的状态机实现，所有对 Repository 的操作都会同步到其他节点，通过 Registry 来操作存储层。

在同步时候，jraft可以直接调用 MetaRepositoryService，实现内部数据一致性。

这里说的是其他节点的同步操作，和上节不同。上节是主动存储节点，这里是被动同步。

put:45, MetaRepositoryService (com.alipay.sofa.registry.server.meta.repository.service)

put:31, RepositoryService (com.alipay.sofa.registry.server.meta.repository)

invokeInterface_L3_L:-1, 1412712349 (java.lang.invoke.LambdaForm$DMH)

reinvoke:-1, 1139092036 (java.lang.invoke.LambdaForm$BMH)

invoker:-1, 403463237 (java.lang.invoke.LambdaForm$MH)

invokeExact_MT:-1, 1568507411 (java.lang.invoke.LambdaForm$MH)

invokeWithArguments:627, MethodHandle (java.lang.invoke)

process:123, Processor (com.alipay.sofa.registry.jraft.processor)

onApply:133, ServiceStateMachine (com.alipay.sofa.registry.jraft.bootstrap)

doApplyTasks:534, FSMCallerImpl (com.alipay.sofa.jraft.core)

doCommitted:503, FSMCallerImpl (com.alipay.sofa.jraft.core)

runApplyTask:431, FSMCallerImpl (com.alipay.sofa.jraft.core)

access$100:72, FSMCallerImpl (com.alipay.sofa.jraft.core)

onEvent:147, FSMCallerImpl$ApplyTaskHandler (com.alipay.sofa.jraft.core)

onEvent:141, FSMCallerImpl$ApplyTaskHandler (com.alipay.sofa.jraft.core)

run:137, BatchEventProcessor (com.lmax.disruptor)

run:748, Thread (java.lang)

1.6 通知data, session

1.6.1 通知操作

前文 "1.3 存储服务" 中有 firePushDataListTask，firePushSessionListTask，最终目的是通知所有data/session有metaServer变动，这里把代码再贴出来温习下。

public class MetaStoreService implements StoreService<MetaNode> {

    @Override

    public NodeChangeResult addNode(MetaNode metaNode) {

            //存放到repository（自动通过jraft同步给集群）

            metaRepositoryService.put(ipAddress, new RenewDecorate(metaNode,

                RenewDecorate.DEFAULT_DURATION_SECS));

            //触发通知(需要通知data/session)，我们说的就是这里

            nodeChangeResult = getNodeChangeResult();

            firePushDataListTask(nodeChangeResult, "addMetaNode");

            firePushSessionListTask(nodeChangeResult, "addMetaNode");

    }

}

firePushDataListTask 和 firePushSessionListTask 进而会往taskListenerManager发送消息，

private void firePushDataListTask(NodeChangeResult nodeChangeResult, String nodeOperate) {

    TaskEvent taskEvent = new TaskEvent(nodeChangeResult, TaskType.DATA_NODE_CHANGE_PUSH_TASK);

    taskEvent.setAttribute(Constant.PUSH_NEED_CONFIRM_KEY, false);

    taskEvent.setAttribute(Constant.PUSH_TARGET_TYPE, NodeType.DATA);

    taskEvent.setAttribute(Constant.PUSH_TARGET_OPERATOR_TYPE, nodeOperate);

    taskListenerManager.sendTaskEvent(taskEvent);

}

private void firePushSessionListTask(NodeChangeResult nodeChangeResult, String nodeOperate) {

    //notify all session node

    TaskEvent taskEvent = new TaskEvent(nodeChangeResult, TaskType.DATA_NODE_CHANGE_PUSH_TASK);

    taskEvent.setAttribute(Constant.PUSH_TARGET_TYPE, NodeType.SESSION);

    taskEvent.setAttribute(Constant.PUSH_TARGET_OPERATOR_TYPE, nodeOperate);

    taskListenerManager.sendTaskEvent(taskEvent);

}

此处是触发通知，于是data 节点, session节点会得到通知。

如何通知data节点, session节点？这里用到了Listener，即调用了 DefaultTaskListenerManager # sendTaskEvent。

堆栈如下：

sendTaskEvent:45, DefaultTaskListenerManager (com.alipay.sofa.registry.task.listener)

firePushDataListTask:355, MetaStoreService (com.alipay.sofa.registry.server.meta.store)

addNode:127, MetaStoreService (com.alipay.sofa.registry.server.meta.store)

addNode:54, MetaStoreService (com.alipay.sofa.registry.server.meta.store)

register:52, MetaServerRegistry (com.alipay.sofa.registry.server.meta.registry)

registerCurrentNode:203, RaftExchanger (com.alipay.sofa.registry.server.meta.remoting)

access$200:56, RaftExchanger (com.alipay.sofa.registry.server.meta.remoting)

startProcess:105, RaftExchanger$1 (com.alipay.sofa.registry.server.meta.remoting)

lambda$onLeaderStart$2:234, ServiceStateMachine (com.alipay.sofa.registry.jraft.bootstrap)

run:-1, 50941601 (com.alipay.sofa.registry.jraft.bootstrap.ServiceStateMachine$$Lambda$192)

1.6.2 分发通知消息

TaskListenerManager 是用来分发各种通知消息的类。

public class DefaultTaskListenerManager implements TaskListenerManager {

    @Override

    public void sendTaskEvent(TaskEvent taskEvent) {

        Collection<TaskListener> taskListeners = this.taskListeners.get(taskEvent.getTaskType());

        for (TaskListener taskListener : taskListeners) {

            taskListener.handleEvent(taskEvent);

        }

    }

}

在 taskListenerManager.addTaskListener(taskListener); 中注册了很多处理消息的handler，从下面运行变量的文字中可以看出其逻辑意义。

this.taskListeners = {ArrayListMultimap@7221}

 expectedValuesPerKey = 3

 map = {HashMap@7227}  size = 4

  {TaskEvent$TaskType@7235} "RECEIVE_STATUS_CONFIRM_NOTIFY_TASK" -> {ArrayList@7236}  size = 1

   key = {TaskEvent$TaskType@7235} "RECEIVE_STATUS_CONFIRM_NOTIFY_TASK"

   value = {ArrayList@7236}  size = 1

    0 = {ReceiveStatusConfirmNotifyTaskListener@7248}

  {TaskEvent$TaskType@7237} "PERSISTENCE_DATA_CHANGE_NOTIFY_TASK" -> {ArrayList@7238}  size = 1

   key = {TaskEvent$TaskType@7237} "PERSISTENCE_DATA_CHANGE_NOTIFY_TASK"

   value = {ArrayList@7238}  size = 1

    0 = {PersistenceDataChangeNotifyTaskListener@7254}

  {TaskEvent$TaskType@7239} "SESSION_NODE_CHANGE_PUSH_TASK" -> {ArrayList@7240}  size = 1

   key = {TaskEvent$TaskType@7239} "SESSION_NODE_CHANGE_PUSH_TASK"

   value = {ArrayList@7240}  size = 1

    0 = {SessionNodeChangePushTaskListener@7252}

  {TaskEvent$TaskType@7241} "DATA_NODE_CHANGE_PUSH_TASK" -> {ArrayList@7242}  size = 1

   key = {TaskEvent$TaskType@7241} "DATA_NODE_CHANGE_PUSH_TASK"

   value = {ArrayList@7242}  size = 1

    0 = {DataNodeChangePushTaskListener@7250}

于是在 sendTaskEvent 之中会调用相关的消息handler来进行处理。

1.6.3 异步处理消息

具体处理消息，我们举例如下：

DataNodeChangePushTaskListener 就是用来处理DataNode相关消息的类。

public class DataNodeChangePushTaskListener implements TaskListener {

    private TaskDispatcher<String, MetaServerTask> dataSingleTaskDispatcher;

    private TaskDispatcher<String, MetaServerTask> sessionSingleTaskDispatcher;

    @Override

    public void handleEvent(TaskEvent event) {

        NodeType nodeType = (NodeType) event.getAttribute(Constant.PUSH_TARGET_TYPE);

        switch (nodeType) {

            case SESSION:

                MetaServerTask sessionNodeChangePushTask = new DataNodeChangePushTask(

                    NodeType.SESSION, metaServerConfig);

                sessionNodeChangePushTask.setTaskEvent(event);

                sessionSingleTaskDispatcher.dispatch(sessionNodeChangePushTask.getTaskId(),

                    sessionNodeChangePushTask, sessionNodeChangePushTask.getExpiryTime());

                break;

            case DATA:

                MetaServerTask dataNodeChangePushTask = new DataNodeChangePushTask(NodeType.DATA,

                    metaServerConfig);

                dataNodeChangePushTask.setTaskEvent(event);

                dataSingleTaskDispatcher.dispatch(dataNodeChangePushTask.getTaskId(),

                    dataNodeChangePushTask, dataNodeChangePushTask.getExpiryTime());

                break;

            default:

                break;

        }

    }

}

我们可以看到，TaskDispatcher 是一个分发异步消息，随之通过TaskExecutors进行异步操作的类。

假如是DataNode，最后会调用到 DataNodeChangePushTask，其是由 DataNodeSingleTaskProcessor 来执行。

其调用栈如下：

pushDataNodes:73, DataNodeServiceImpl (com.alipay.sofa.registry.server.meta.node.impl)

execute:86, DataNodeChangePushTask (com.alipay.sofa.registry.server.meta.task)

process:41, DataNodeSingleTaskProcessor (com.alipay.sofa.registry.server.meta.task.processor)

process:32, DataNodeSingleTaskProcessor (com.alipay.sofa.registry.server.meta.task.processor)

run:136, TaskExecutors$WorkerRunnable (com.alipay.sofa.registry.task.batcher)

run:748, Thread (java.lang)

具体到 DataNodeChangePushTask 来进行与各个DataNode进行交互的操作。

public class DataNodeChangePushTask extends AbstractMetaServerTask {

    private final SessionNodeService sessionNodeService;

    private final DataNodeService    dataNodeService;

    final private MetaServerConfig   metaServerConfig;

    final private NodeType           nodeType;

    private NodeChangeResult         nodeChangeResult;

    private Boolean                  confirm;

    private String                   confirmNodeIp;

    private Map<String, DataNode>    targetNodes;

    @Override

    public void execute() {

        switch (nodeType) {

            case SESSION:

                sessionNodeService.pushDataNodes(nodeChangeResult);

                break;

            case DATA:

                dataNodeService

                    .pushDataNodes(nodeChangeResult, targetNodes, confirm, confirmNodeIp);

                break;

            default:

                break;

        }

    }

}

以DataNodeServiceImpl为例，可以看到最后调用了 dataNodeExchanger 完成了节点间通讯。

public class DataNodeServiceImpl implements DataNodeService {

    @Autowired

    private NodeExchanger         dataNodeExchanger;

    @Autowired

    private StoreService          dataStoreService;

    @Autowired

    private AbstractServerHandler dataConnectionHandler;

    @Override

    public void pushDataNodes(NodeChangeResult nodeChangeResult, Map<String, DataNode> targetNodes,

                              boolean confirm, String confirmNodeIp) {

        if (nodeChangeResult != null) {

            List<Throwable> exceptions = new ArrayList<>();

            NodeConnectManager nodeConnectManager = getNodeConnectManager();

            Collection<InetSocketAddress> connections = nodeConnectManager.getConnections(null);

            // add register confirm

            StoreService storeService = ServiceFactory.getStoreService(NodeType.DATA);

            DataCenterNodes dataCenterNodes = storeService.getDataCenterNodes();

            Map<String, DataNode> registeredNodes = dataCenterNodes.getNodes();

            for (InetSocketAddress address : connections) {

                try {

                    if (targetNodes != null && !targetNodes.isEmpty()) {

                        if (!targetNodes.keySet().contains(address.getAddress().getHostAddress())) {

                            continue;

                        }

                    } else {

                        if (!registeredNodes.keySet().contains(

                            address.getAddress().getHostAddress())) {

                            continue;

                        }

                    }

                    Request<NodeChangeResult> nodeChangeRequestRequest = new Request<NodeChangeResult>() {

                        @Override

                        public NodeChangeResult getRequestBody() {

                            return nodeChangeResult;

                        }

                        @Override

                        public URL getRequestUrl() {

                            return new URL(address);

                        }

                    };

                    // 节点间通讯

                    Response response = dataNodeExchanger.request(nodeChangeRequestRequest);

                    if (confirm) {

                        Object result = response.getResult();

                        if (result instanceof CommonResponse) {

                            CommonResponse genericResponse = (CommonResponse) result;

                            if (genericResponse.isSuccess()) {

                                confirmStatus(address, confirmNodeIp);

                            }

                        }

                    }

                }

            }

        }

    }

}

1.6.4 另一种产生通知方式

ExecutorManager的pushNodeListChange会定期检查，如果有必要，则产生通知。

对应后文的 DataConfirmStatusService 节点变更事件。

scheduler.schedule(

        new TimedSupervisorTask("CheckDataNodeListChangePush", scheduler, checkNodeListChangePushExecutor,

                metaServerConfig.getSchedulerCheckNodeListChangePushTimeout(), TimeUnit.SECONDS,

                metaServerConfig.getSchedulerCheckNodeListChangePushExpBackOffBound(),

                () -> metaServerRegistry.pushNodeListChange(NodeType.DATA)),

        metaServerConfig.getSchedulerCheckNodeListChangePushFirstDelay(), TimeUnit.SECONDS);

其堆栈是：

pushNodeListChange:278, DataStoreService (com.alipay.sofa.registry.server.meta.store)

pushNodeListChange:103, MetaServerRegistry (com.alipay.sofa.registry.server.meta.registry)

lambda$startScheduler$4:158, ExecutorManager (com.alipay.sofa.registry.server.meta.executor)

run:-1, 758751909 (com.alipay.sofa.registry.server.meta.executor.ExecutorManager$$Lambda$202)

call:511, Executors$RunnableAdapter (java.util.concurrent)

run$$$capture:266, FutureTask (java.util.concurrent)

run:-1, FutureTask (java.util.concurrent)

比如

public class DataStoreService implements StoreService<DataNode> {

    @Override

    public void pushNodeListChange() {

        NodeOperator<DataNode> fireNode;

        if ((fireNode = dataConfirmStatusService.peekConfirmNode()) != null) {

            NodeChangeResult nodeChangeResult = getNodeChangeResult();

            Map<String, Map<String, DataNode>> map = nodeChangeResult.getNodes();

            Map<String, DataNode> addNodes = map.get(nodeConfig.getLocalDataCenter());

            if (addNodes != null) {

                Map<String, DataNode> previousNodes = dataConfirmStatusService.putExpectNodes(

                    fireNode.getNode(), addNodes);

                if (!previousNodes.isEmpty()) {

                    // 产生通知

                    firePushDataListTask(fireNode, nodeChangeResult, previousNodes, true);

                }

            }

            // 产生通知

            firePushSessionListTask(nodeChangeResult, fireNode.getNodeOperate().toString());

        }

    }

}

0x02 节点注册

2.1 DataApplication

2.1.1 DataConnectionHandler

当一个DataApplication启动，首先DataConnectionHandler会响应。

connected:40, DataConnectionHandler (com.alipay.sofa.registry.server.meta.remoting.connection)

onEvent:69, ConnectionEventAdapter (com.alipay.sofa.registry.remoting.bolt)

onEvent:44, ConnectionEventListener (com.alipay.remoting)

run:201, ConnectionEventHandler$1 (com.alipay.remoting)

runWorker:1149, ThreadPoolExecutor (java.util.concurrent)

run:624, ThreadPoolExecutor$Worker (java.util.concurrent)

run:748, Thread (java.lang)

感觉只是设置了一下。

public class DataConnectionHandler extends AbstractServerHandler implements NodeConnectManager {

    private Map<String/*connectId*/, InetSocketAddress> connections = new ConcurrentHashMap<>();

    @Override

    public void connected(Channel channel) throws RemotingException {

        super.connected(channel);

        addConnection(channel);

    }

}

2.1.2 DataNodeHandler

然后是 DataNodeHandler 会响应。

reply:43, DataNodeHandler (com.alipay.sofa.registry.server.meta.remoting.handler)

handleRequest:54, SyncUserProcessorAdapter (com.alipay.sofa.registry.remoting.bolt)

dispatchToUserProcessor:239, RpcRequestProcessor (com.alipay.remoting.rpc.protocol)

doProcess:145, RpcRequestProcessor (com.alipay.remoting.rpc.protocol)

run:366, RpcRequestProcessor$ProcessTask (com.alipay.remoting.rpc.protocol)

runWorker:1149, ThreadPoolExecutor (java.util.concurrent)

run:624, ThreadPoolExecutor$Worker (java.util.concurrent)

run:748, Thread (java.lang)

具体代码如下。

public class DataNodeHandler extends AbstractServerHandler<DataNode> {

    @Autowired

    private Registry            metaServerRegistry;

    @Override

    public Object reply(Channel channel, DataNode dataNode) {

        NodeChangeResult nodeChangeResult;

        nodeChangeResult = metaServerRegistry.register(dataNode);

        return nodeChangeResult;

    }

}

接下来是调用Store服务进行添加节点。

@Override

public NodeChangeResult register(Node node) {

    StoreService storeService = ServiceFactory.getStoreService(node.getNodeType());

    return storeService.addNode(node);

}

代码来到 DataStoreService，依然是调用 RepositoryService 来进行节点的注册和存储。

public class DataStoreService implements StoreService<DataNode> {

    @Autowired

    private TaskListenerManager                                taskListenerManager;

    @RaftReference(uniqueId = "dataServer")

    private RepositoryService<String, RenewDecorate<DataNode>> dataRepositoryService;

    @RaftReference(uniqueId = "dataServer")

    private NodeConfirmStatusService<DataNode>                 dataConfirmStatusService;

    @Override

    public NodeChangeResult addNode(DataNode dataNode) {

        NodeChangeResult nodeChangeResult;

        String ipAddress = dataNode.getNodeUrl().getIpAddress();

        write.lock();

        try {

            dataRepositoryService.put(ipAddress, new RenewDecorate(dataNode,

                RenewDecorate.DEFAULT_DURATION_SECS));

            renew(dataNode, 30);

            nodeChangeResult = getNodeChangeResult();

            dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.ADD);

        } finally {

            write.unlock();

        }

        return nodeChangeResult;

    }

}

dataConfirmStatusService.putConfirmNode 同时会存储一个变更事件到队列中，主要用于数据推送，消费处理。

@Override

public void putConfirmNode(DataNode node, DataOperator nodeOperate) {

	expectNodesOrders.put(new NodeOperator(node, nodeOperate));

}

2.1.3 DataConfirmStatusService

DataConfirmStatusService 也是注解了RaftService，这说明是一个由 Raft 协议进行同步的存储。

以下的这些存储结构会被同步。

expectNodesOrders 用来存储节点变更事件；
expectNodes 用来存储变更事件需要确认的节点，也就是说 NodeOperator 只有得到了其他节点的确认，才会从 expectNodesOrders 移除；
snapShotFileNames 是快照文件名；

@RaftService(uniqueId = "dataServer")

public class DataConfirmStatusService extends AbstractSnapshotProcess

                                      implements NodeConfirmStatusService<DataNode> {

	private ConcurrentHashMap<DataNode/*node*/, Map<String/*ipAddress*/, DataNode>> expectNodes             = new ConcurrentHashMap<>();

	private BlockingQueue<NodeOperator>                                               expectNodesOrders       = new LinkedBlockingQueue();

	private Set<String>   snapShotFileNames 

	@Override

  public void putConfirmNode(DataNode node, DataOperator nodeOperate) {

      expectNodesOrders.put(new NodeOperator(node, nodeOperate));

  }    

  public NodeOperator<DataNode> peekConfirmNode() {

      return expectNodesOrders.peek();

  }

}

2.1.4 消费

事件存储到 BlockingQueue expectNodesOrders 里，哪里去消费呢？看源码发现，并不是想象中的使用一个线程阻塞的读。

在ExecutorManager中会启动一个定时任务，轮询该队列有没有数据。即周期性的调用Registry#pushNodeListChange方法，获取队列的头节点并消费。Data 和 Session 各对应一个任务。

具体流程如下图所示：

首先获取队列（expectNodesOrders）头节点，如果为Null直接返回；
获取当前数据中心的节点列表，并存储到确认表（expectNodes）；
提交节点变更推送任务（firePushXxListTask)；
处理任务，即调用 XxNodeService 的 pushXxxNode 方法，即通过 ConnectionHandler 获取所有的节点连接，发送节点列表；
收到回复后，如果需要确认，则会调用StoreService#confirmNodeStatus 方法，将该节点从expectNodes中移除；
待所有的节点从 expectNodes 中移除，则将此次操作从 expectNodesOrders 移除，处理完毕；

具体可以细化如下。

2.1.4.1 定时提交节点变更推送任务

本小节对应上面步骤的前三步。

比如关于Data 定时任务如下，里面定期调用了MetaServerRegistry 的 pushNodeListChange函数来处理任务：

scheduler.schedule(

        new TimedSupervisorTask("CheckDataNodeListChangePush", scheduler, checkNodeListChangePushExecutor,

                metaServerConfig.getSchedulerCheckNodeListChangePushTimeout(), TimeUnit.SECONDS,

                metaServerConfig.getSchedulerCheckNodeListChangePushExpBackOffBound(),

                () -> metaServerRegistry.pushNodeListChange(NodeType.DATA)),

        metaServerConfig.getSchedulerCheckNodeListChangePushFirstDelay(), TimeUnit.SECONDS);

MetaServerRegistry 中的 pushNodeListChange 定义如下：

@Override

public void pushNodeListChange(NodeType nodeType) {

    StoreService storeService = ServiceFactory.getStoreService(nodeType);

    if (storeService != null) {

        storeService.pushNodeListChange();

    }

}

DataStoreService 这里做了以下工作。

首先获取队列（expectNodesOrders）头节点，如果为Null直接返回；
获取当前数据中心的节点列表，并存储到确认表（expectNodes）；
提交节点变更推送任务（firePushXxListTask)；

@Override

public void pushNodeListChange() {

    NodeOperator<DataNode> fireNode;

    //  首先获取队列（expectNodesOrders）头节点，如果为Null直接返回；

    if ((fireNode = dataConfirmStatusService.peekConfirmNode()) != null) {

        NodeChangeResult nodeChangeResult = getNodeChangeResult();

        Map<String, Map<String, DataNode>> map = nodeChangeResult.getNodes();

        Map<String, DataNode> addNodes = map.get(nodeConfig.getLocalDataCenter());

        if (addNodes != null) {

            // 获取当前数据中心的节点列表，并存储到确认表（expectNodes）；

            Map<String, DataNode> previousNodes = dataConfirmStatusService.putExpectNodes(

                fireNode.getNode(), addNodes);

            // 提交节点变更推送任务（firePushXxListTask)；

            if (!previousNodes.isEmpty()) {

                firePushDataListTask(fireNode, nodeChangeResult, previousNodes, true);

            }

        }

        // 提交节点变更推送任务（firePushXxListTask)；

        firePushSessionListTask(nodeChangeResult, fireNode.getNodeOperate().toString());

    }

}

这样就把任务提交到了Task任务。

2.1.4.2 定时异步处理任务

本小节对应上面步骤的后三步。

下面就是定时异步处理任务。

处理任务，即调用 XxNodeService 的 pushXxxNode 方法，即通过 ConnectionHandler 获取所有的节点连接，发送节点列表；
收到回复后，如果需要确认，则会调用StoreService#confirmNodeStatus 方法，将该节点从expectNodes中移除；
待所有的节点从 expectNodes 中移除，则将此次操作从 expectNodesOrders 移除，处理完毕；

public class DataNodeServiceImpl implements DataNodeService {

    @Override

    public void pushDataNodes(NodeChangeResult nodeChangeResult, Map<String, DataNode> targetNodes,

                              boolean confirm, String confirmNodeIp) {

        if (nodeChangeResult != null) {

            NodeConnectManager nodeConnectManager = getNodeConnectManager();

            Collection<InetSocketAddress> connections = nodeConnectManager.getConnections(null);

            // add register confirm

            StoreService storeService = ServiceFactory.getStoreService(NodeType.DATA);

            DataCenterNodes dataCenterNodes = storeService.getDataCenterNodes();

            Map<String, DataNode> registeredNodes = dataCenterNodes.getNodes();

            // 通过 ConnectionHandler 获取所有的节点连接，发送节点列表；

            for (InetSocketAddress address : connections) {

                try {

                    Request<NodeChangeResult> nodeChangeRequestRequest = new Request<NodeChangeResult>() {

                        @Override

                        public NodeChangeResult getRequestBody() {

                            return nodeChangeResult;

                        }

                        @Override

                        public URL getRequestUrl() {

                            return new URL(address);

                        }

                    };

                    // 发送节点列表；

                    Response response = dataNodeExchanger.request(nodeChangeRequestRequest);

                    if (confirm) {

                        Object result = response.getResult();

                        if (result instanceof CommonResponse) {

                            CommonResponse genericResponse = (CommonResponse) result;

                            if (genericResponse.isSuccess()) {

                             // 收到回复后，如果需要确认，则会调用`StoreService#confirmNodeStatus` 方法，将该节点从expectNodes中移除；

                                confirmStatus(address, confirmNodeIp);

                            }

                        }

                    }

                }

            }

        }

    }

}

其堆栈如下：

pushDataNodes:73, DataNodeServiceImpl (com.alipay.sofa.registry.server.meta.node.impl)

execute:86, DataNodeChangePushTask (com.alipay.sofa.registry.server.meta.task)

process:41, DataNodeSingleTaskProcessor (com.alipay.sofa.registry.server.meta.task.processor)

process:32, DataNodeSingleTaskProcessor (com.alipay.sofa.registry.server.meta.task.processor)

run:136, TaskExecutors$WorkerRunnable (com.alipay.sofa.registry.task.batcher)

run:748, Thread (java.lang)

2.1.4.3 确认步骤

进一步细化上文提到的 “确认步骤”

private void confirmStatus(InetSocketAddress address, String confirmNodeIp) {

    String ipAddress = address.getAddress().getHostAddress();

    dataStoreService.confirmNodeStatus(ipAddress, confirmNodeIp);

}

然后是 DataStoreService

收到回复后，如果需要确认，则会调用StoreService#confirmNodeStatus 方法，将该节点从expectNodes中移除；
待所有的节点从 expectNodes 中移除，则将此次操作从 expectNodesOrders 移除，处理完毕；

@Override

public void confirmNodeStatus(String ipAddress, String confirmNodeIp) {

    NodeOperator<DataNode> fireNode = dataConfirmStatusService.peekConfirmNode();

    if (fireNode != null) {

        String fireNodeIp = fireNode.getNode().getIp();

        Map<String/*ipAddress*/, DataNode> waitNotifyNodes = dataConfirmStatusService

            .getExpectNodes(fireNode.getNode());

        if (waitNotifyNodes != null) {

            Set<String> removeIp = getRemoveIp(waitNotifyNodes.keySet());

            removeIp.add(ipAddress);

            // 将该节点从expectNodes中移除；

            waitNotifyNodes = dataConfirmStatusService.removeExpectConfirmNodes(

                fireNode.getNode(), removeIp);

            if (waitNotifyNodes.isEmpty()) {

                //all node be notified,or some disconnect node be evict

                try {

                    // 待所有的节点从 expectNodes 中移除，则将此次操作从 expectNodesOrders 移除，处理完毕；

                    if (null != dataConfirmStatusService

                        .removeExpectNodes((dataConfirmStatusService.pollConfirmNode())

                            .getNode())) {

                        //add init status must notify

                        if (fireNode.getNodeOperate() == DataOperator.ADD) {

                            notifyConfirm(fireNode.getNode());

                        }

                    }

                }

            }

        } else {

            try {

                //wait node not exist,

                dataConfirmStatusService.pollConfirmNode();

            }

        }

    }

}

2.1.4.4 总结消费流程

对于消费流程，总结如下图：

 +-------------------+        +-----------+         +------------------------+                                       +----------------+

 |ServiceStateMachine+------> | Processor | +-----> |DataConfirmStatusService|                                       |     loop       |

 +-------------------+        +-----------+         +-----+------------------+                                       v                |

                                                          |                                                   +------+------+         |

                                                          | put                                               |TaskExecutors|         |

                                                          v                                                   +------+------+         |

                                                 +--------+--------+   remo^e                                        |                |

                            +------------------> |expectNodesOrders| <-------+                                       |                |

                            | loop               +--------+--------+         |                                       v                |

                            |                             |                  |                          +------------+--------------+ |

                            |                             | peekConfirmNode  |                          |DataNodeSingleTaskProcessor| |

                            |                             |                  |                          +------------+--------------+ |

                            |                             |                  |                                       |                |

+---------------------------+--------------------+        v                  |                                       |                |

|               ExecutorManager                  |   +----+---+              |                                       v                |

| +--------------------------------------------+ |   |fireNode|              |                            +----------+-----------+    |

| |           TimedSuper^isorTask              | |   +----+---+              |                            |DataNodeChangePushTask|    |

| |  +---------------------------------------+ | |        |                  |                            +----------+-----------+    |

| |  | metaServerRegistry.pushNodeListChange | | |        | putExpectNodes   |                                       |                |

| |  +---------------------------------------+ | |        |                  |                                       |                |

| +--------------------------------------------+ |        v                  |                                       v                |

+------------------------------------------------+ +------+------+   remo^e  |                             +---------+---------+      |

                            ^                      | expectNodes | <---------+------------------<--------+ |DataNodeServiceImpl|      |

                            |                      +------+------+                                       | +---------+---------+      |

                            |                             |                                              |           |                |

                            |                             v                                              |           |                |

                            |               +-------------+--------+                                     |           v                |

                            |               | firePushDataListTask +-----------------------------------> |    +------+-------+        |

                            |               +-------------+--------+ taskListenerManager.sendTaskEvent   |    | StoreService |        |

                            |                             |                                              |    +------+-------+        ^

                            |                             v                                              |           |                |

                            |               +-------------+-----------+                                  |           v                |

                            ^<--------------+ firePushSessionListTask +--------------------------------> |     +-----+-------+        |

                                            +-------------------------+taskListenerManager.sendTaskEvent +-----+confirmStatus|        |

                                                                                                               +-----+-------+        |

                                                                                                                     |                |

                                                                                                                     >----------------+

手机上参见如下

2.1.5 总结

数据节点注册流程总结如下图。

                                                                                                       +----------------------------+

                                                              put        +----------------------+      | Map(String, NodeRepository)|

                                                            +----------> |dataRepositoryService +----> | registry                   |

                                                            |            +----------------------+      +----------------------------+

                                                            |

                 register                   addNode         |

+----------------+      +------------------+        +-------+--------+

| DataNodeHandler+----->+metaServerRegistry+------->+DataStoreService|                                       TimedSupervisorTask

+----------------+      +------------------+        +-------+--------+

                                                            |            +------------------------+          +------------------+

                                                            +----------> |dataConfirmStatusService| +------> |pushNodeListChange|

                                                          putConfirmNode +------------------------+          +------------------+

手机上参见如下

2.2 SessionNodeHandler

Session节点的注册和Data节点几乎类似。

SessionNodeHandler 定义如下：

public class SessionNodeHandler extends AbstractServerHandler<SessionNode> {

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

    @Autowired

    private Registry            metaServerRegistry;

    @Override

    public Object reply(Channel channel, SessionNode sessionNode) {

        NodeChangeResult nodeChangeResult;

        try {

            nodeChangeResult = metaServerRegistry.register(sessionNode);

        }

        return nodeChangeResult;

    }

}

然后是进行业务处理的SessionStoreService。

public class SessionStoreService implements StoreService<SessionNode> {

    @Override

    public NodeChangeResult addNode(SessionNode sessionNode) {

        write.lock();

        try {

            String ipAddress = sessionNode.getNodeUrl().getIpAddress();

            sessionRepositoryService.put(ipAddress, new RenewDecorate(sessionNode,

                RenewDecorate.DEFAULT_DURATION_SECS));

            sessionVersionRepositoryService.checkAndUpdateVersions(nodeConfig.getLocalDataCenter(),

                System.currentTimeMillis());

            renew(sessionNode, 30);

            sessionConfirmStatusService.putConfirmNode(sessionNode, DataOperator.ADD);

        } finally {

            write.unlock();

        }

        return dataStoreService.getNodeChangeResult();

    }

}

然后是SessionRepositoryService。

注意，这里map的key是ip，这是与Data不一致的地方。

@RaftService(uniqueId = "sessionServer")

public class SessionRepositoryService extends AbstractSnapshotProcess

                                                                     implements

                                                                     RepositoryService<String, RenewDecorate<SessionNode>> {

    /**

     * session node store

     */

    private ConcurrentHashMap<String/*ipAddress*/, RenewDecorate<SessionNode>> registry          = new ConcurrentHashMap<>();

    @Override

    public RenewDecorate<SessionNode> put(String ipAddress, RenewDecorate<SessionNode> sessionNode,

                                          Long currentTimeMillis) {

        try {

            RenewDecorate oldRenewDecorate = registry.get(ipAddress);

            registry.put(ipAddress, sessionNode);

        }

        return sessionNode;

    }

}

由上节可知，DataServer 和 SessionServer 都有处理节点注册请求的 Handler。注册行为由 Registry 完成。

比如 metaServerHandlers。

@Bean(name = "metaServerHandlers")

public Collection<AbstractServerHandler> metaServerHandlers() {

    Collection<AbstractServerHandler> list = new ArrayList<>();

    list.add(metaConnectionHandler());

    list.add(getNodesRequestHandler());

    return list;

}

0x03 注册信息续约

3.1 关键类 RenewDecorate

节点注册的时候，节点信息被 RenewDecorate 包装起来了，这个就是实现注册信息续约和驱逐的关键，该类定义如下：

public class RenewDecorate<T> implements Serializable {

    public static final int DEFAULT_DURATION_SECS = 15;

    private T               renewal;  // 节点对象封装

    private long            beginTimestamp; // 注册事件

    private volatile long   lastUpdateTimestamp; // 续约时间

    private long            duration; // 超时时间

    public boolean isExpired() {

        return System.currentTimeMillis() > lastUpdateTimestamp + duration;

    }

    public void renew() {

        lastUpdateTimestamp = System.currentTimeMillis() + duration;

    }

    public void renew(long durationSECS) {

        lastUpdateTimestamp = System.currentTimeMillis() + durationSECS * 1000;

    }

}

该对象为注册节点信息，附加了注册时间、上次续约时间、过期时间。

续约操作就是修改lastUpdateTimestamp，是否过期就是判断System.currentTimeMillis() - lastUpdateTimestamp > duration 是否成立，成立则认为节点超时进行驱逐。

3.2 执行路径

可以看到，renew可以从多条执行路径调用。

路径一：

renew:70, RenewDecorate (com.alipay.sofa.registry.server.meta.store)

replace:176, DataRepositoryService (com.alipay.sofa.registry.server.meta.repository.service)

replace:45, DataRepositoryService (com.alipay.sofa.registry.server.meta.repository.service)

replace:39, RepositoryService (com.alipay.sofa.registry.server.meta.repository)

invokeInterface_L3_L:-1, 1889671331 (java.lang.invoke.LambdaForm$DMH)

reinvoke:-1, 171310548 (java.lang.invoke.LambdaForm$BMH)

invoker:-1, 559310456 (java.lang.invoke.LambdaForm$MH)

invokeExact_MT:-1, 426124479 (java.lang.invoke.LambdaForm$MH)

invokeWithArguments:627, MethodHandle (java.lang.invoke)

process:123, Processor (com.alipay.sofa.registry.jraft.processor)

onApply:133, ServiceStateMachine (com.alipay.sofa.registry.jraft.bootstrap)

doApplyTasks:534, FSMCallerImpl (com.alipay.sofa.jraft.core)

doCommitted:503, FSMCallerImpl (com.alipay.sofa.jraft.core)

runApplyTask:431, FSMCallerImpl (com.alipay.sofa.jraft.core)

access$100:72, FSMCallerImpl (com.alipay.sofa.jraft.core)

onEvent:147, FSMCallerImpl$ApplyTaskHandler (com.alipay.sofa.jraft.core)

onEvent:141, FSMCallerImpl$ApplyTaskHandler (com.alipay.sofa.jraft.core)

run:137, BatchEventProcessor (com.lmax.disruptor)

run:748, Thread (java.lang)

路径二：

renew:171, DataStoreService (com.alipay.sofa.registry.server.meta.store)

addNode:111, DataStoreService (com.alipay.sofa.registry.server.meta.store)

addNode:60, DataStoreService (com.alipay.sofa.registry.server.meta.store)

register:52, MetaServerRegistry (com.alipay.sofa.registry.server.meta.registry)

reply:43, DataNodeHandler (com.alipay.sofa.registry.server.meta.remoting.handler)

reply:32, DataNodeHandler (com.alipay.sofa.registry.server.meta.remoting.handler)

handleRequest:54, SyncUserProcessorAdapter (com.alipay.sofa.registry.remoting.bolt)

dispatchToUserProcessor:239, RpcRequestProcessor (com.alipay.remoting.rpc.protocol)

doProcess:145, RpcRequestProcessor (com.alipay.remoting.rpc.protocol)

run:366, RpcRequestProcessor$ProcessTask (com.alipay.remoting.rpc.protocol)

runWorker:1149, ThreadPoolExecutor (java.util.concurrent)

run:624, ThreadPoolExecutor$Worker (java.util.concurrent)

run:748, Thread (java.lang)

比如在 DataStoreService 之中

@Override

public NodeChangeResult addNode(DataNode dataNode) {

    NodeChangeResult nodeChangeResult;

    String ipAddress = dataNode.getNodeUrl().getIpAddress();

    write.lock();

    try {

        dataRepositoryService.put(ipAddress, new RenewDecorate(dataNode,

            RenewDecorate.DEFAULT_DURATION_SECS));

        renew(dataNode, 30);  // 续约

        nodeChangeResult = getNodeChangeResult();

        dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.ADD);

    } finally {

        write.unlock();

    }

    return nodeChangeResult;

}

进而调用renew

@Override

public void renew(DataNode dataNode, int duration) {

    write.lock();

    try {

        String ipAddress = dataNode.getNodeUrl().getIpAddress();

        RenewDecorate renewer = dataRepositoryService.get(ipAddress);

        if (renewer == null) {

            addNode(dataNode); // 新增

        } else {

            // 续约

            if (duration > 0) {

                dataRepositoryService.replace(ipAddress, new RenewDecorate(dataNode, duration));

            } else {

                dataRepositoryService.replace(ipAddress, new RenewDecorate(dataNode,

                    RenewDecorate.DEFAULT_DURATION_SECS));

            }

        }

    } finally {

        write.unlock();

    }

}

因为是Raft对Repository进行数据一致性维护，所以 dataRepositoryService.replace 会被 Proxy 替换，进而来到了ProxyHandler。

@Override

public Object invoke(Object proxy, Method method, Object[] args) {

    try {

        ProcessRequest request = new ProcessRequest();

        request.setMethodArgSigs(createParamSignature(method.getParameterTypes()));

        request.setMethodName(method.getName());

        request.setMethodArgs(args);

        request.setServiceName(serviceId);

        if (Processor.getInstance().isLeaderReadMethod(method)) {

            return doInvokeMethod(request);

        }

        return client.sendRequest(request);

    }

}

当 Raft 来到 服务端，堆栈如下：

replace:165, DataRepositoryService (com.alipay.sofa.registry.server.meta.repository.service)

replace:45, DataRepositoryService (com.alipay.sofa.registry.server.meta.repository.service)

replace:39, RepositoryService (com.alipay.sofa.registry.server.meta.repository)

invokeInterface_L3_L:-1, 1177311202 (java.lang.invoke.LambdaForm$DMH)

reinvoke:-1, 960369282 (java.lang.invoke.LambdaForm$BMH)

invoker:-1, 198860519 (java.lang.invoke.LambdaForm$MH)

invokeExact_MT:-1, 2035225037 (java.lang.invoke.LambdaForm$MH)

invokeWithArguments:627, MethodHandle (java.lang.invoke)

process:123, Processor (com.alipay.sofa.registry.jraft.processor)

onApply:133, ServiceStateMachine (com.alipay.sofa.registry.jraft.bootstrap)

doApplyTasks:534, FSMCallerImpl (com.alipay.sofa.jraft.core)

doCommitted:503, FSMCallerImpl (com.alipay.sofa.jraft.core)

runApplyTask:431, FSMCallerImpl (com.alipay.sofa.jraft.core)

access$100:72, FSMCallerImpl (com.alipay.sofa.jraft.core)

onEvent:147, FSMCallerImpl$ApplyTaskHandler (com.alipay.sofa.jraft.core)

onEvent:141, FSMCallerImpl$ApplyTaskHandler (com.alipay.sofa.jraft.core)

run:137, BatchEventProcessor (com.lmax.disruptor)

run:748, Thread (java.lang)

具体renew代码如下，这样就最终完成了更新：

@Override

public RenewDecorate<DataNode> replace(String ipAddress, RenewDecorate<DataNode> dataNode,

                                       Long currentTimeMillis) {

    try {

        String dataCenter = dataNode.getRenewal().getDataCenter();

        NodeRepository<DataNode> dataNodeRepository = registry.get(dataCenter);

        if (dataNodeRepository != null) {

            Map<String/*ipAddress*/, RenewDecorate<DataNode>> dataNodes = dataNodeRepository

                .getNodeMap();

            RenewDecorate<DataNode> oldRenewDecorate = dataNodes.get(ipAddress);

            if (oldRenewDecorate != null && oldRenewDecorate.getRenewal() != null) {

                oldRenewDecorate.setRenewal(dataNode.getRenewal());

                oldRenewDecorate.renew();

            }

        }

        return dataNode;

    }

}

具体如下图

+---------------------------------------+        +---------------------------------------------+

| +------------------------------+      |        |       +----------------------------------+  |

| | +----------------+  registry |Client|        | Server| +----------------------+registry |  |

| | |DataStoreService|           |      |        |       | | DataRepositoryService|         |  |

| | +-----+----------+           |      |        |       | +---------+------------+         |  |

| |       | replace              |      |        |       |           ^  replace             |  |

| |       |                      |      |        |       |           |                      |  |

| |       v                      |      |        |       |    +------+----+                 |  |

| | +-----+--------------------+ |      |        |       |    | Processor |                 |  |

| | |DataRepositoryService stub| |      |        |       |    +------+----+                 |  |

| | +-----+--------------------+ |      |        |       |           ^  onApply             |  |

| |       |                      |      |        |       |           |                      |  |

| |       v                      |      |        |       |   +-------+------+               |  |

| |     +-+---+                  |      |        |       |   | StateMachine |               |  |

| |     |Proxy|                  |      |        |       |   +-------+------+               |  |

| |     +-+---+                  |      |        |       |           ^  process             |  |

| |       | invoke               |      |        |       |           |                      |  |

| |       v                      |      |        |       |           |                      |  |

| |  +----+-------+              |      |        |       |    +------+------+               |  |

| |  |ProxyHandler|              |      |        |       |    |FSMCallerImpl|               |  |

| |  +----+-------+              |      |        |       |    +------+------+               |  |

| |       | sendRequest          |      |        |       |           ^                      |  |

| |       v                      |      |        |       |           |  received            |  |

| |   +---+------+               |      |        |       |           |                      |  |

| |   |RaftClient|               |      |        |       |   +-----------------+            |  |

| |   +----------+               |      | network|       |   |RaftServerHandler|            |  |

| |                              | +-------------------> |   +-----------------+            |  |

| +------------------------------+      |        |       +----------------------------------+  |

|                                       |        |                                             |

+---------------------------------------+        +---------------------------------------------+

手机上如下

3.3 ReNewNodesRequestHandler

和注册一样，续约请求的处理 Handler 为ReNewNodesRequestHandler，最终交由 StoreService 进行续约操作。另外一点，续约的时候如果没有查询到注册节点，会触发节点注册的操作。

在初始化时候，就设置了ReNew handler。

        @Bean(name = "sessionServerHandlers")

        public Collection<AbstractServerHandler> sessionServerHandlers() {

            Collection<AbstractServerHandler> list = new ArrayList<>();

            list.add(sessionConnectionHandler());

            list.add(sessionNodeHandler());

            list.add(renewNodesRequestHandler()); // 注册到Server handler

            list.add(getNodesRequestHandler());

            list.add(fetchProvideDataRequestHandler());

            return list;

        }

        @Bean(name = "dataServerHandlers")

        public Collection<AbstractServerHandler> dataServerHandlers() {

            Collection<AbstractServerHandler> list = new ArrayList<>();

            list.add(dataConnectionHandler());

            list.add(getNodesRequestHandler());

            list.add(dataNodeHandler());

            list.add(renewNodesRequestHandler()); // 注册到Server handler

            list.add(fetchProvideDataRequestHandler());

            return list;

        }

具体定义如下：

public class RenewNodesRequestHandler extends AbstractServerHandler<RenewNodesRequest> {

    @Autowired

    private Registry            metaServerRegistry;

    @Override

    public Object reply(Channel channel, RenewNodesRequest renewNodesRequest) {

        Node renewNode = null;

        renewNode = renewNodesRequest.getNode();

        metaServerRegistry.renew(renewNode, renewNodesRequest.getDuration());

        return null;

    }

    @Override

    public Class interest() {

        return RenewNodesRequest.class;

    }

    @Override

    public HandlerType getType() {

        return HandlerType.PROCESSER;

    }

}

0x04 驱除

驱出的操作是由定时任务完成，MetaServer 在启动时会启动多个定时任务，详见ExecutorManager#startScheduler，其中一个任务会调用Registry#evict，其实现为遍历存储的 Map, 获得过期的列表，调用StoreService#removeNodes方法，将他们从 Repository 中移除，这个操作也会触发变更通知。该任务默认每3秒执行一次。

4.1 配置

public void startScheduler() {

    scheduler.schedule(new TimedSupervisorTask("HeartbeatCheck", scheduler, heartbeatCheckExecutor,

                    metaServerConfig.getSchedulerHeartbeatTimeout(), TimeUnit.SECONDS,

                    metaServerConfig.getSchedulerHeartbeatExpBackOffBound(), () -> metaServerRegistry.evict()),

            metaServerConfig.getSchedulerHeartbeatFirstDelay(), TimeUnit.SECONDS);

}

4.2 驱除

这里就是遍历各种StoreService，获取其中的过期节点，然后进行驱除。

public class MetaServerRegistry implements Registry<Node> {

    @Override

    public void evict() {

        for (NodeType nodeType : NodeType.values()) {

            StoreService storeService = ServiceFactory.getStoreService(nodeType);

            if (storeService != null) {

                Collection<Node> expiredNodes = storeService.getExpired();

                if (expiredNodes != null && !expiredNodes.isEmpty()) {

                    storeService.removeNodes(expiredNodes);

                }

            }

        }

    }

}

具体又会继续调用 DataStoreService。

@Override

public void removeNodes(Collection<DataNode> nodes) {

    write.lock();

    try {

        if (nodes != null && !nodes.isEmpty()) {

            for (DataNode dataNode : nodes) {

                String ipAddress = dataNode.getNodeUrl().getIpAddress();

                RenewDecorate<DataNode> dataNodeRemove = dataRepositoryService

                    .remove(ipAddress);

                if (dataNodeRemove != null) {

                    dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.REMOVE);

                }

            }

        }

    } finally {

        write.unlock();

    }

}

最后调用到 DataRepositoryService

@Override

public RenewDecorate<DataNode> remove(Object key, Long currentTimeMillis) {

    try {

        String ipAddress = (String) key;

        String dataCenter = nodeConfig.getLocalDataCenter();

        NodeRepository<DataNode> dataNodeRepository = registry.get(dataCenter);

        if (dataNodeRepository != null) {

            Map<String/*ipAddress*/, RenewDecorate<DataNode>> dataNodes = dataNodeRepository

                .getNodeMap();

            if (dataNodes != null) {

                RenewDecorate<DataNode> oldRenewDecorate = dataNodes.remove(ipAddress);

                dataNodeRepository.setVersion(currentTimeMillis);

                return oldRenewDecorate;

            }

        }

    }

}

0x05 节点列表查询

Data，Meta, Session Server 都提供 getNodesRequestHandler ，用于处理查询当前节点列表的请求，其本质上从底层存储 Repository 读取数据返回。返回的结果的具体结构见 NodeChangeResult 类，包含各个数据中心的节点列表以及版本号。

public class NodeChangeResult<T extends Node> implements Serializable {

    private final NodeType                                               nodeType;

    private Map<String/*dataCenter id*/, Map<String /*ipAddress*/, T>> nodes;

    private Long                                                         version;

    private Map<String/*dataCenter*/, Long /*version*/>                dataCenterListVersions;

    /** local dataCenter id */

    private String                                                       localDataCenter;

}

5.1 配置

Data，Meta，Session Server 都提供 getNodesRequestHandler。具体如下：

@Bean(name = "sessionServerHandlers")

public Collection<AbstractServerHandler> sessionServerHandlers() {

    Collection<AbstractServerHandler> list = new ArrayList<>();

    list.add(sessionConnectionHandler());

    list.add(sessionNodeHandler());

    list.add(renewNodesRequestHandler());

    list.add(getNodesRequestHandler()); // 做了配置

    list.add(fetchProvideDataRequestHandler());

    return list;

}

@Bean(name = "dataServerHandlers")

public Collection<AbstractServerHandler> dataServerHandlers() {

    Collection<AbstractServerHandler> list = new ArrayList<>();

    list.add(dataConnectionHandler());

    list.add(getNodesRequestHandler()); // 做了配置

    list.add(dataNodeHandler());

    list.add(renewNodesRequestHandler());

    list.add(fetchProvideDataRequestHandler());

    return list;

}

@Bean(name = "metaServerHandlers")

public Collection<AbstractServerHandler> metaServerHandlers() {

    Collection<AbstractServerHandler> list = new ArrayList<>();

    list.add(metaConnectionHandler());

    list.add(getNodesRequestHandler()); // 做了配置

    return list;

}

getNodesRequestHandler的Bean配置生成如下：

@Configuration

public static class MetaServerRemotingConfiguration {

        @Bean

        public AbstractServerHandler getNodesRequestHandler() {

            return new GetNodesRequestHandler();

        }

}

5.2 响应handler

NodesRequest 是通过Bolt来响应消息。

public class GetNodesRequestHandler extends AbstractServerHandler<GetNodesRequest> {

    @Autowired

    private Registry            metaServerRegistry;

    @Override

    public Object reply(Channel channel, GetNodesRequest getNodesRequest) {

        NodeChangeResult nodeChangeResult;

        try {

            nodeChangeResult = metaServerRegistry.getAllNodes(getNodesRequest.getNodeType());

        }

        return nodeChangeResult;

    }

}

堆栈为：

getNodeChangeResult:188, MetaStoreService (com.alipay.sofa.registry.server.meta.store)

getAllNodes:96, MetaServerRegistry (com.alipay.sofa.registry.server.meta.registry)

reply:44, GetNodesRequestHandler (com.alipay.sofa.registry.server.meta.remoting.handler)

reply:33, GetNodesRequestHandler (com.alipay.sofa.registry.server.meta.remoting.handler)

handleRequest:54, SyncUserProcessorAdapter (com.alipay.sofa.registry.remoting.bolt)

dispatchToUserProcessor:239, RpcRequestProcessor (com.alipay.remoting.rpc.protocol)

doProcess:145, RpcRequestProcessor (com.alipay.remoting.rpc.protocol)

run:366, RpcRequestProcessor$ProcessTask (com.alipay.remoting.rpc.protocol)

runWorker:1149, ThreadPoolExecutor (java.util.concurrent)

run:624, ThreadPoolExecutor$Worker (java.util.concurrent)

run:748, Thread (java.lang)

具体如下图所示：

                      +-------------------------+

                      | SyncUserProcessorAdapter|

                      +-----------+-------------+

                                  |

                                  |

                                  v  handleRequest

                       +----------+-----------+

                       |GetNodesRequestHandler|

                       +----------+-----------+

                                  |

                                  |

                                  v  getAllNodes

                         +--------+---------+

                         |MetaServerRegistry|

                         +--------+---------+

                                  |

                                  |

                                  v  getNodeChangeResult

                          +-------+--------+

                          |MetaStoreService|

                          +----+------+----+

                               |      |

                   +-----------+      +------------+

                   |                               |

      getNodeMap   v                               v  getNodeMap

+------------------+-----------+        +----------+-------------------+

|dataCenter, metaNodeRepository|  ...   |dataCenter, metaNodeRepository|

+------------------------------+        +------------------------------+

我们可以再具体深入下。

5.3 Registry操作

Registry 只是简单调用StoreService。

public class MetaServerRegistry implements Registry<Node> {

    @Override

    public NodeChangeResult getAllNodes(NodeType nodeType) {

        StoreService storeService = ServiceFactory.getStoreService(nodeType);

        return storeService.getNodeChangeResult();

    }

}

5.4 StoreService操作

Service 会遍历数据中心，获取对应的Node列表以及version，最终返回。

public class MetaStoreService implements StoreService<MetaNode> {

    @Override

    public NodeChangeResult getNodeChangeResult() {

        NodeChangeResult nodeChangeResult = new NodeChangeResult(NodeType.META);

        String localDataCenter = nodeConfig.getLocalDataCenter();

        Map<String/*dataCenter*/, NodeRepository> metaRepositoryMap = metaRepositoryService.getNodeRepositories();

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

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

        metaRepositoryMap.forEach((dataCenter, metaNodeRepository) -> {

            if (localDataCenter.equalsIgnoreCase(dataCenter)) {

                nodeChangeResult.setVersion(metaNodeRepository.getVersion());

            }

            versionMap.put(dataCenter, metaNodeRepository.getVersion());

            Map<String, RenewDecorate<MetaNode>> dataMap = metaNodeRepository.getNodeMap();

            Map<String, MetaNode> newMap = new ConcurrentHashMap<>();

            dataMap.forEach((ip, dataNode) -> newMap.put(ip, dataNode.getRenewal()));

            pushNodes.put(dataCenter, newMap);

        });

        nodeChangeResult.setLocalDataCenter(localDataCenter);

        nodeChangeResult.setNodes(pushNodes);

        nodeChangeResult.setDataCenterListVersions(versionMap);

        return nodeChangeResult;

    }

}

最后结果是：

nodeChangeResult =

 nodeType = {Node$NodeType@7190} "DATA"

 nodes = {ConcurrentHashMap@8267}  size = 1

  "DefaultDataCenter" -> {ConcurrentHashMap@8276}  size = 0

   key = "DefaultDataCenter"

   value = {ConcurrentHashMap@8276}  size = 0

 version = {Long@8268} 1601126414990

 dataCenterListVersions = {ConcurrentHashMap@8269}  size = 1

  "DefaultDataCenter" -> {Long@8268} 1601126414990

   key = "DefaultDataCenter"

   value = {Long@8268} 1601126414990

 localDataCenter = "DefaultDataCenter"

  value = {char[17]@8280}

  hash = 761435552

5.5 节点变更时的数据同步

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

本文提到了很多Raft相关信息，下篇文章就具体探究下究竟MetaServer是如何使用 Raft。