序

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

EurekaServerAutoConfiguration

@Configuration
@Import(EurekaServerInitializerConfiguration.class)
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
@EnableConfigurationProperties({ EurekaDashboardProperties.class,InstanceRegistryProperties.class })
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {//......@Beanpublic 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.java

    @PostConstruct@Overridepublic void initialize() throws Exception {logger.info("Initializing ...");peerEurekaNodes.start();registry.init(peerEurekaNodes);logger.info("Initialized");}@PreDestroy@Overridepublic 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();

start

PeerEurekaNodes.start

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

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

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，然后剔除自己的url

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 changereturn;}// Remove peers no long availableList<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 peersif (!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创建

    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);}
复制代码

添加的时候会创建新的PeerEurekaNode

registry.init(peerEurekaNodes)

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

    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、还有初始化RemoteRegionRegistry

shutdown

registry.shutdown()

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

    /*** Perform all cleanup and shutdown operations.*/@Overridepublic 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.*/@Overridepublic 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