服务注册发现：Eureka源码剖析（11）核心机制之Eureka Server集群机制

class: DefaultEurekaServerContext

@PostConstruct
@Override
public void initialize() {
    logger.info("Initializing ...");
    // 启动eureka集群，这个集群是每个eureka server节点都维护的一组内存对象。
    // 他的作用是每当本节点发生变更，比如有服务实例注册、下线、心跳续约等事件发生时，会修改这组对象
    // 然后这组对象负责将修改同步到对应的eureka server节点上去。
    peerEurekaNodes.start();
    try {
        // 初始化服务注册表
        // 这里其实是将所有的eureka server节点上的注册表都抓取过来，合并一下
        // 然后存放到自己的本地注册表中去
        registry.init(peerEurekaNodes);
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    logger.info("Initialized");
}

class: PeerEurekaNodes
    
public void start() {
    // 搞了一个单线程的线程池 用来跑eureka集群节点的更新任务
    taskExecutor = Executors.newSingleThreadScheduledExecutor(
        new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                thread.setDaemon(true);
                return thread;
            }
        }
    );
    try {
        // 这里是更新自己内存中的eureka集群，
        // 对新的eureka节点进行创建实例，然后添加到集群进去
        // 对已经无效的eureka 节点清除出这个集群中
        // 这里被Spring Cloud Netflix Eureka拿来做eureka集群的在线扩缩容

        // 更新eureka server集群信息
        updatePeerEurekaNodes(resolvePeerUrls());
        Runnable peersUpdateTask = new Runnable() {
            @Override
            public void run() {
                try {
                    updatePeerEurekaNodes(resolvePeerUrls());
                } catch (Throwable e) {
                    logger.error("Cannot update the replica Nodes", e);
                }

            }
        };
        taskExecutor.scheduleWithFixedDelay(
            peersUpdateTask,
            serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
            serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
            TimeUnit.MILLISECONDS
        );
    } catch (Exception e) {
        throw new IllegalStateException(e);
    }
    for (PeerEurekaNode node : peerEurekaNodes) {
        logger.info("Replica node URL:  {}", node.getServiceUrl());
    }
}

class: PeerEurekaNodes
    
protected void updatePeerEurekaNodes(List<String> newPeerUrls) {

    // 这里toShutdown就是需要缩容的eureka server节点
    // 实现很简单，就是以最新的服务地址列表为准，除此之外的都需要干掉
    Set<String> toShutdown = new HashSet<>(peerEurekaNodeUrls);
    toShutdown.removeAll(newPeerUrls);

    // 这里toAdd就是需要扩容的eureka server节点
    // 实现很简单，就是拿新增的服务地址列表
    Set<String> toAdd = new HashSet<>(newPeerUrls);
    toAdd.removeAll(peerEurekaNodeUrls);

    // Remove peers no long available
    List<PeerEurekaNode> newNodeList = new ArrayList<>(peerEurekaNodes);

    // 这里处理缩容的逻辑
    // 遍历现在的集群里的节点，删除缩容的节点，并且关停缩容的节点
    // 关停缩容的节点 其实没啥特别的 就是释放一些线程池资源而已
    if (!toShutdown.isEmpty()) {
        logger.info("Removing no longer available peer nodes {}", toShutdown);
        int i = 0;
        while (i < newNodeList.size()) {
            PeerEurekaNode eurekaNode = newNodeList.get(i);
            if (toShutdown.contains(eurekaNode.getServiceUrl())) {
                newNodeList.remove(i);
                eurekaNode.shutDown();
            } else {
                i++;
            }
        }
    }

    // 这里处理扩容的逻辑
    // 遍历需要扩容的节点，创建之，加入集群
    // Add new peers
    if (!toAdd.isEmpty()) {
        logger.info("Adding new peer nodes {}", toAdd);
        for (String peerUrl : toAdd) {
            newNodeList.add(createPeerEurekaNode(peerUrl));
        }
    }

    this.peerEurekaNodes = newNodeList;
    this.peerEurekaNodeUrls = new HashSet<>(newPeerUrls);
}