3、Eureka-源码-Eureka Server启动入口调试_enableeurekaclient 代码入口

springcloud eureka分为两个模块eureka-server、eureka-client。

一、eureka入口分析

经过下面的几步可以知道eureka的入口类是EurekaServerAutoConfiguration。

@SpringBootApplication
//1、关于eureka入口
@EnableEurekaServer
public class EurekaServerApplication {

	public static void main(String[] args) {
		SpringApplication.run(EurekaServerApplication.class, args);
	}
}

//2、点击入口类一路追踪，发现其注入一个空的标记类Marker
@Configuration
public class EurekaServerMarkerConfiguration {

	@Bean
	public Marker eurekaServerMarkerBean() {
		return new Marker();
	}

	class Marker {
	}
}

/*3、搜索标记类Marker找到EurekaServerAutoConfiguration 
  3.1、ConditionalOnBean注释判断是否已经实例化Marker，然后加载EurekaServerAutoConfiguration
*/
/**
 * @author Gunnar Hillert
 * @author Biju Kunjummen
 */
@Configuration
//注入EurekaServerInitializerConfiguration到spring ioc容器
@Import(EurekaServerInitializerConfiguration.class)
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
//注入配置类
@EnableConfigurationProperties({ EurekaDashboardProperties.class,
		InstanceRegistryProperties.class })
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

二、入口类EurekaServerAutoConfiguration做了什么？

从读代码我们的成果是： eureka启动拉取别的节点配置次数为5、服务的生命周期为60s、检查服务活跃状态的定时器为30s、找到集群数据同步方法peerEurekaNodes

2.1、声明收集集群中其他配置的次数

@Configuration
	protected static class EurekaServerConfigBeanConfiguration {
		@Bean
        //保证容器实例化且只有一个实例
		@ConditionalOnMissingBean
		public EurekaServerConfig eurekaServerConfig(EurekaClientConfig clientConfig) {
			EurekaServerConfigBean server = new EurekaServerConfigBean();
			if (clientConfig.shouldRegisterWithEureka()) {
				// 声明eureka启动的时候收集集群其他节点配置信息的次数
				server.setRegistrySyncRetries(5);
			}
			return server;
		}
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.2、初始化eurekaServer、eurekaClient

//1、初始化eurekaServer、eurekaClient入口	
@Bean
public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
			ServerCodecs serverCodecs) {
		this.eurekaClient.getApplications(); // force initialization
		return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig,
				serverCodecs, this.eurekaClient,
				this.instanceRegistryProperties.getExpectedNumberOfRenewsPerMin(),
				this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
}

//2、初始化服务注册环境变量
protected AbstractInstanceRegistry(EurekaServerConfig serverConfig, EurekaClientConfig clientConfig, ServerCodecs serverCodecs) {
        this.serverConfig = serverConfig;
        this.clientConfig = clientConfig;
        this.serverCodecs = serverCodecs;
        //2.1、最近下线的服务队列
        this.recentCanceledQueue = new CircularQueue<Pair<Long, String>>(1000);
        //2.2、最近上线的服务队列
        this.recentRegisteredQueue = new CircularQueue<Pair<Long, String>>(1000);
        //2.3、最少服务心跳次数默认心跳小于 numberOfRenewsPerMinThreshold 时，并且开启自动保护模式开关( eureka.enableSelfPreservation = true ) 时，触发自动保护机制，不再自动过期租约
        this.renewsLastMin = new MeasuredRate(1000 * 60 * 1);
        
        //2.4、启动后30s后，后续每隔30秒执行一次
        this.deltaRetentionTimer.schedule(getDeltaRetentionTask(),
                serverConfig.getDeltaRetentionTimerIntervalInMs(),
                serverConfig.getDeltaRetentionTimerIntervalInMs());
}

//3、服务注册存放的队列数据结构
private ConcurrentLinkedQueue<RecentlyChangedItem> recentlyChangedQueue = new ConcurrentLinkedQueue<RecentlyChangedItem>();

//4、移除服务注册队列里面60秒未更新的记录
private TimerTask getDeltaRetentionTask() {
        return new TimerTask() {

            @Override
            public void run() {
                Iterator<RecentlyChangedItem> it = recentlyChangedQueue.iterator();
                //4.1、迭代服务队列如果服务最后更新时间是60s以前移除
                while (it.hasNext()) {
                    if (it.next().getLastUpdateTime() <
                            System.currentTimeMillis() - serverConfig.getRetentionTimeInMSInDeltaQueue()) {
                        it.remove();
                    } else {
                        break;
                    }
                }
            }

        };
}

//5、集群节点数据同步
@Bean
@ConditionalOnMissingBean
public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
			ServerCodecs serverCodecs) {
        //5.1、集群节点同步逻辑
		return new PeerEurekaNodes(registry, this.eurekaServerConfig,
				this.eurekaClientConfig, serverCodecs, this.applicationInfoManager);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

三、集群节点创建入口类PeerEurekaNodes

读代码收获 直达守护线程概念及作用、eureka节点自己更新流程

//1、PeerEurekaNodes入口
public void start() {
        /*1.1、为什么在线程池设置守护线程？目的烦那个防止线程池不退出，当线程池有任务主线程退出了他也不会退出,
         JVM官方解释：当 JVM 中不存在任何一个正在运行的非守护线程时，则 JVM 进程即会退出。
        */
        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);
                    }

                }
            };
            //1.2、定时器延时10分钟执行一次，后续10分钟执行一次节点数据同步
            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());
        }
    }


//3、eureka节点状态更新
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;
        }

        //3.0、当前和历史节点做比较，移除历史和当前不一致的节点
        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;
        }

        //3.1、关闭掉队eureka集群节点
        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) {
                //4、创建集群节点
                newNodeList.add(createPeerEurekaNode(peerUrl));
            }
        }
        
        //4.1、当节点初始化后留下当前节点快照，为后续移除节点做比较
        this.peerEurekaNodes = newNodeList;
        this.peerEurekaNodeUrls = new HashSet<>(newPeerUrls);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

四、集群节点创建过程类PeerEurekaNode

此类太难，先放弃，后续在搞。

4.1、集群节点创建入口方法

//1、集群节点创建入口
protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
        //http线程池
        HttpReplicationClient replicationClient = JerseyReplicationClient.createReplicationClient(serverConfig, serverCodecs, peerEurekaNodeUrl);
        String targetHost = hostFromUrl(peerEurekaNodeUrl);
        if (targetHost == null) {
            targetHost = "host";
        }
        return new PeerEurekaNode(registry, targetHost, peerEurekaNodeUrl, replicationClient, serverConfig);
}
1
2
3
4
5
6
7
8
9
10

4.2、http连接池维护

//ApacheHttpClientConnectionCleaner.java
//连接池30s不使用销毁
public ApacheHttpClientConnectionCleaner(ApacheHttpClient4 apacheHttpClient, final long connectionIdleTimeout) {
        this.apacheHttpClient = apacheHttpClient;
        this.eurekaConnCleaner.scheduleWithFixedDelay(
                new Runnable() {
                    @Override
                    public void run() {
                        cleanIdle(connectionIdleTimeout);
                    }
                },
                HTTP_CONNECTION_CLEANER_INTERVAL_MS,
                HTTP_CONNECTION_CLEANER_INTERVAL_MS,
                TimeUnit.MILLISECONDS
        );

        MonitorConfig.Builder monitorConfigBuilder = MonitorConfig.builder("Eureka-Connection-Cleaner-Time");
        executionTimeStats = new BasicTimer(monitorConfigBuilder.build());
        cleanupFailed = new BasicCounter(MonitorConfig.builder("Eureka-Connection-Cleaner-Failure").build());
        try {
            Monitors.registerObject(this);
        } catch (Exception e) {
            logger.error("Unable to register with servo.", e);
     }
}

//连接池销毁
public void cleanIdle(long delayMs) {
        Stopwatch start = executionTimeStats.start();
        try {
            apacheHttpClient.getClientHandler().getHttpClient()
                    .getConnectionManager()
                    .closeIdleConnections(delayMs, TimeUnit.SECONDS);
        } catch (Throwable e) {
            logger.error("Cannot clean connections", e);
            cleanupFailed.increment();
        } finally {
            if (null != start) {
                start.stop();
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42