聊聊spring cloud的DefaultEurekaServerContext

时间 2019-11-24

标签聊聊 spring cloud defaulteurekaservercontext 栏目 Spring 繁體版

原文原文链接

序

本文主要研究一下spring cloud的DefaultEurekaServerContextjava

EurekaServerAutoConfiguration

@Configuration
@Import(EurekaServerInitializerConfiguration.class)
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
@EnableConfigurationProperties({ EurekaDashboardProperties.class,
		InstanceRegistryProperties.class })
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
	//......
	@Bean
	public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
			PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
		return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
				registry, peerEurekaNodes, this.applicationInfoManager);
	}
	//......
}
复制代码

DefaultEurekaServerContext

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/DefaultEurekaServerContext.javanode

@PostConstruct
    @Override
    public void initialize() throws Exception {
        logger.info("Initializing ...");
        peerEurekaNodes.start();
        registry.init(peerEurekaNodes);
        logger.info("Initialized");
    }

    @PreDestroy
    @Override
    public void shutdown() throws Exception {
        logger.info("Shutting down ...");
        registry.shutdown();
        peerEurekaNodes.shutdown();
        logger.info("Shut down");
    }
复制代码

实例化后的时候执行peerEurekaNodes.start();以及registry.init(peerEurekaNodes); 销毁以前执行registry.shutdown();以及peerEurekaNodes.shutdown();git

start

PeerEurekaNodes.start

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/cluster/PeerEurekaNodes.javagithub

public void start() {
        taskExecutor = Executors.newSingleThreadScheduledExecutor(
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                        thread.setDaemon(true);
                        return thread;
                    }
                }
        );
        try {
            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());
        }
    }
复制代码

这里首先执行updatePeerEurekaNodes，以后注册定时任务去定时触发updatePeerEurekaNodes，时间间隔为erverConfig.getPeerEurekaNodesUpdateIntervalMs()spring

resolvePeerUrls

/**
     * Resolve peer URLs.
     *
     * @return peer URLs with node's own URL filtered out */ protected List<String> resolvePeerUrls() { InstanceInfo myInfo = applicationInfoManager.getInfo(); String zone = InstanceInfo.getZone(clientConfig.getAvailabilityZones(clientConfig.getRegion()), myInfo); List<String> replicaUrls = EndpointUtils .getDiscoveryServiceUrls(clientConfig, zone, new EndpointUtils.InstanceInfoBasedUrlRandomizer(myInfo)); int idx = 0; while (idx < replicaUrls.size()) { if (isThisMyUrl(replicaUrls.get(idx))) { replicaUrls.remove(idx); } else { idx++; } } return replicaUrls; } 复制代码

首先经过resolvePeerUrls来获取replicaUrls，这里获取的是健康的url，而后剔除本身的urlbash

updatePeerEurekaNodes

/**
     * Given new set of replica URLs, destroy {@link PeerEurekaNode}s no longer available, and
     * create new ones.
     *
     * @param newPeerUrls peer node URLs; this collection should have local node's URL filtered out */ protected void updatePeerEurekaNodes(List<String> newPeerUrls) { if (newPeerUrls.isEmpty()) { logger.warn("The replica size seems to be empty. Check the route 53 DNS Registry"); return; } Set<String> toShutdown = new HashSet<>(peerEurekaNodeUrls); toShutdown.removeAll(newPeerUrls); Set<String> toAdd = new HashSet<>(newPeerUrls); toAdd.removeAll(peerEurekaNodeUrls); if (toShutdown.isEmpty() && toAdd.isEmpty()) { // No change return; } // 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); } 复制代码

主要是跟原始或上次更新的peerEurekaNodeUrls对比，移除掉不健康的节点，移除的时候会调用PeerEurekaNode的shutdown方法，添加的时候经过createPeerEurekaNode建立app

protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
        HttpReplicationClient replicationClient = JerseyReplicationClient.createReplicationClient(serverConfig, serverCodecs, peerEurekaNodeUrl);
        String targetHost = hostFromUrl(peerEurekaNodeUrl);
        if (targetHost == null) {
            targetHost = "host";
        }
        return new PeerEurekaNode(registry, targetHost, peerEurekaNodeUrl, replicationClient, serverConfig);
    }
复制代码

添加的时候会建立新的PeerEurekaNodedom

registry.init(peerEurekaNodes)

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/PeerAwareInstanceRegistryImpl.javaide

public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
        this.numberOfReplicationsLastMin.start();
        this.peerEurekaNodes = peerEurekaNodes;
        initializedResponseCache();
        scheduleRenewalThresholdUpdateTask();
        initRemoteRegionRegistry();

        try {
            Monitors.registerObject(this);
        } catch (Throwable e) {
            logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
        }
    }
复制代码

这里初始化ResponseCache、调度RenewalThresholdUpdateTask、还有初始化RemoteRegionRegistryui

shutdown

registry.shutdown()

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/PeerAwareInstanceRegistryImpl.java

/**
     * Perform all cleanup and shutdown operations.
     */
    @Override
    public void shutdown() {
        try {
            DefaultMonitorRegistry.getInstance().unregister(Monitors.newObjectMonitor(this));
        } catch (Throwable t) {
            logger.error("Cannot shutdown monitor registry", t);
        }
        try {
            peerEurekaNodes.shutdown();
        } catch (Throwable t) {
            logger.error("Cannot shutdown ReplicaAwareInstanceRegistry", t);
        }
        numberOfReplicationsLastMin.stop();

        super.shutdown();
    }
复制代码

这里主要是调用peerEurekaNodes.shutdown()，还有super的shutdown

super.shutdown

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/AbstractInstanceRegistry.java

/**
     * Perform all cleanup and shutdown operations.
     */
    @Override
    public void shutdown() {
        deltaRetentionTimer.cancel();
        evictionTimer.cancel();
        renewsLastMin.stop();
    }
复制代码

主要是关闭一些计时器

peerEurekaNodes.shutdown()

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/cluster/PeerEurekaNodes.java

public void shutdown() {
        taskExecutor.shutdown();
        List<PeerEurekaNode> toRemove = this.peerEurekaNodes;

        this.peerEurekaNodes = Collections.emptyList();
        this.peerEurekaNodeUrls = Collections.emptySet();

        for (PeerEurekaNode node : toRemove) {
            node.shutDown();
        }
    }
复制代码

除了系统关闭会调用外，该eureka node被认为不健康的时候，被剔除时，也会调用shutdown方法

PeerEurekaNode.shutdown()

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/cluster/PeerEurekaNode.java

/**
     * Shuts down all resources used for peer replication.
     */
    public void shutDown() {
        batchingDispatcher.shutdown();
        nonBatchingDispatcher.shutdown();
    }
复制代码

主要是关闭dispatcher

小结

EurekaServerContext主要注册了bean初始化及销毁时执行的操做。初始化时，启动peerNodes，而后初始化registry；销毁时，关闭registry，而后关闭peerNodes。关于peerNodes，主要的是定时任务以erverConfig.getPeerEurekaNodesUpdateIntervalMs()时间间隔去定时触发updatePeerEurekaNodes，而这个操做是跟原始或上次更新的peerEurekaNodeUrls对比，移除掉不健康的节点，添加新的节点若是有的话。移除的时候会调用PeerEurekaNode的shutdown方法，添加新的peerNode的时候经过createPeerEurekaNode建立。

doc

Understanding Eureka Peer to Peer Communication