背景说明

本文针对eureka的源码分析，基于的版本号：

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.14</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
 <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
            <version>3.1.7</version>
        </dependency>

知识点

SmartLifecycle接口

在spring启动后，或者停止前，需要做一些操作，就可以实现该接口。
在spring启动后，需要把应用注册到eureka server，服务停止前，需要从eureka server剔除该应用。实现方式就是实现了SmartLifecycle接口。

public class EurekaAutoServiceRegistration
		implements AutoServiceRegistration, SmartLifecycle, Ordered, SmartApplicationListener {
	@Override
	public void start() {
		// only set the port if the nonSecurePort or securePort is 0 and this.port != 0
		if (this.port.get() != 0) {
			if (this.registration.getNonSecurePort() == 0) {
				this.registration.setNonSecurePort(this.port.get());
			}

			if (this.registration.getSecurePort() == 0 && this.registration.isSecure()) {
				this.registration.setSecurePort(this.port.get());
			}
		}

		// only initialize if nonSecurePort is greater than 0 and it isn't already running
		// because of containerPortInitializer below
		if (!this.running.get() && this.registration.getNonSecurePort() > 0) {

			this.serviceRegistry.register(this.registration);

			this.context.publishEvent(new InstanceRegisteredEvent<>(this, this.registration.getInstanceConfig()));
			this.running.set(true);
		}
	}

	@Override
	public void stop() {
		this.serviceRegistry.deregister(this.registration);
		this.running.set(false);
	}
}

SmartLifecycle 和ApplicationListener功能类似。

@import注解

作用：可以用来动态加载一个Configuration文件，或者动态加载一个Bean
使用场景：当项目启动后，还不确定要加载哪个bean时，就可以使用这种方式。
https://zhuanlan.zhihu.com/p/147025312

spring.factories 文件

文件中，定义的是接口和接口的实现类。
在Spring启动过程中，不同时机会加载不同接口的实现类。
如当添加了@SpringBootApplication注解之后，就会自动加载EnableAutoConfiguration接口的实现类，spring boot的自动配置功能就是这么实现的。

@EnableConfigurationProperties注解

https://www.jianshu.com/p/7f54da1cb2eb
当我们定义一个类，这个类会映射到application.yaml中的配置，但是又不想把这个类声明成一个Bean，就可以通过@EnableConfigurationProperties把Properties类声明成一个bean

在spring boot2.7版本中，@EnableDiscoveryClient 注解已经没用了

即使不使用这个注解，classpath中有了eureka，也会进行服务注册。
如果classpath中已经有了eureka，又想禁用服务注册，怎么办？

spring.cloud.service-registry.auto-registration.enabled = false

spring框架自定义事件通知机制

场景：spring启动时，当应用程序注册到eureka server之后，需要通知DiscoveryClientHealthIndicator（健康指示器），告诉他已经注册完成。实现方式：
（1）监听自定义事件

public class DiscoveryClientHealthIndicator
		implements DiscoveryHealthIndicator, Ordered, ApplicationListener<InstanceRegisteredEvent<?>> {
	@Override
	public void onApplicationEvent(InstanceRegisteredEvent<?> event) {
		if (this.discoveryInitialized.compareAndSet(false, true)) {
			this.log.debug("Discovery Client has been initialized");
		}
	}
}

（2）发送事件：

this.context.publishEvent(new InstanceRegisteredEvent<>(this, this.registration.getInstanceConfig()));

疑问：事件监听程序，运行在哪个线程中？

eureka client 服务注册源码解析

1、/spring-cloud-netflix-eureka-client/3.1.7/spring-cloud-netflix-eureka-client-3.1.7-sources.jar!/META-INF/spring.factories 中定义自动配置类

org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration

2、接着就会注入EurekaClientAutoConfiguration中的bean

	@Bean
	@ConditionalOnBean(AutoServiceRegistrationProperties.class)
	@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
	public EurekaAutoServiceRegistration eurekaAutoServiceRegistration(ApplicationContext context,
			EurekaServiceRegistry registry, EurekaRegistration registration) {
		return new EurekaAutoServiceRegistration(context, registry, registration);
	}

3、在EurekaAutoServiceRegistration中，实现了SmartLifecycle的start、stop接口。因此，应用启动、停止过程中，就会自动进行服务的注册、注销。

	@Override
	public void start() {
		// only initialize if nonSecurePort is greater than 0 and it isn't already running
		// because of containerPortInitializer below
		if (!this.running.get() && this.registration.getNonSecurePort() > 0) {

			this.serviceRegistry.register(this.registration);

			this.context.publishEvent(new InstanceRegisteredEvent<>(this, this.registration.getInstanceConfig()));
			this.running.set(true);
		}
	}

4、关键代码this.serviceRegistry.register，继续追踪，会执行com.netflix.appinfo.ApplicationInfoManager中的setInstanceStatus()方法：

5、通过线程池，调用InstanceInfoReplicator.run()方法：

6、查找可用的eureka server，发送请求，进行实例注册

自问自答

eureka client注册到server时，负载均衡机制是怎样的？如果一个eureka server挂了，会发生什么？

1、选择server instance的逻辑：

	@Override
    public List<AwsEndpoint> getClusterEndpoints() {
        List<AwsEndpoint>[] parts = ResolverUtils.splitByZone(delegate.getClusterEndpoints(), myZone);
        List<AwsEndpoint> myZoneEndpoints = parts[0];
        List<AwsEndpoint> remainingEndpoints = parts[1];
        List<AwsEndpoint> randomizedList = randomizeAndMerge(myZoneEndpoints, remainingEndpoints);
        if (!zoneAffinity) {
            Collections.reverse(randomizedList);
        }

        logger.debug("Local zone={}; resolved to: {}", myZone, randomizedList);

        return randomizedList;
    }

会选择和client同区域的server

2、如果有多个待选instance，当需要向server发送请求时，如果遇到server响应错误（500），会重试，请求下一台server

private final AtomicReference<EurekaHttpClient> delegate = new AtomicReference<>();

int endpointIdx = 0;
for (int retry = 0; retry < numberOfRetries; retry++) {
    // 记录当前使用的http client（代表当前正在使用的eureka server）
    EurekaHttpClient currentHttpClient = delegate.get();
    EurekaEndpoint currentEndpoint = null;
    // 如果没有可用的server，就进行查询。如果已经找到了，下次就直接使用
    if (currentHttpClient == null) {
        if (candidateHosts == null) {
            candidateHosts = getHostCandidates();
            if (candidateHosts.isEmpty()) {
                throw new TransportException("There is no known eureka server; cluster server list is empty");
            }
        }
        if (endpointIdx >= candidateHosts.size()) {
            throw new TransportException("Cannot execute request on any known server");
        }
        // 如果当前没有可用实例（或者当前实例请求失败了），就取下一个server，进行尝试
        currentEndpoint = candidateHosts.get(endpointIdx++);
        currentHttpClient = clientFactory.newClient(currentEndpoint);
    }

    try {
        EurekaHttpResponse<R> response = requestExecutor.execute(currentHttpClient);
        if (serverStatusEvaluator.accept(response.getStatusCode(), requestExecutor.getRequestType())) {
            delegate.set(currentHttpClient);
            if (retry > 0) {
                logger.info("Request execution succeeded on retry #{}", retry);
            }
            return response;
        }
        logger.warn("Request execution failure with status code {}; retrying on another server if available", response.getStatusCode());
    } catch (Exception e) {
        logger.warn("Request execution failed with message: {}", e.getMessage());  // just log message as the underlying client should log the stacktrace
    }

    // Connection error or 5xx from the server that must be retried on another server
    // 如果使用当前server请求失败了，就把currentHttpClient置为null，下次循环时，重新查找可用server
    delegate.compareAndSet(currentHttpClient, null);
    if (currentEndpoint != null) {
        quarantineSet.add(currentEndpoint);
    }
}
throw new TransportException("Retry limit reached; giving up on completing the request");

eureka集群是点对点架构（没有master），会存在数据写入冲突的问题吗？

eureka架构，满足分布式原理cap中的ap，不满足一致性。理由：
1、往eureka中写入的数据（注册一个服务），要过一段时间才能查询到。写入之后不能立即查询到，所以不是一致性。是最终一致性
2、在多实例eureka集中中，当client注册到一个节点之后，它会异步地把数据发送到其他所有节点。由于是异步的，如果数据写入A节点，从B节点查询时肯定是有延迟的。

不会存在写入冲突。理由：
1、同一个key（client实例信息），正常情况下只会写入第一个instance （eureka.client.serviceUrl.defaultZone中第一个server）。
2、如果遇到第一个server挂了，会尝试请求下一个server。此时，同一个key会写入不同的instance。由于服务注册的数据写入都是幂等的（多次insert效果一样、没有递增操作），因为不存在数据冲突的问题。

如何快速阅读java源码？

带着问题：“eureka server不同实例之间如何复制数据？”，来分享下如果快速找到相关的源码？

1、看源码最开始最懵逼的是，不知道相关逻辑的代码在哪里？

途径一：把spring日志改为debug模式，通过关键日志，推测相关代码逻辑在哪个文件

途径二：通过wireshark抓包，查看服务间都调用了哪些http接口，然后在代码中搜索相关接口。

比如客户端注册的是eureka 8761端口的实例，从结果看，很快8762、8763两个实例都有该app的信息了。
那么，我就通过wireshark监听和端口8762之间的通信：

看到在请求8762端口上的POST /eureka/peerreplication/batch/ HTTP/1.1接口，参数是：

看起来，8761实例通过调用上面的接口，发app信息同步到了8762

搜索接口名称，找到了相关代码之后，继续分析：

private void replicateToPeers(Action action, String appName, String id,
                                  InstanceInfo info /* optional */,
                                  InstanceStatus newStatus /* optional */, boolean isReplication) {
        Stopwatch tracer = action.getTimer().start();
        try {
            if (isReplication) {
                numberOfReplicationsLastMin.increment();
            }
            // If it is a replication already, do not replicate again as this will create a poison replication
            if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {
                return;
            }

            for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {
                // If the url represents this host, do not replicate to yourself.
                if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {
                    continue;
                }
                replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
            }
        } finally {
            tracer.stop();
        }
    }

另外，这里的复制是异步进行的。具体操作方式（记得不太清了）：
1、建一个阻塞队列。当有新客户端注册时，把实例信息发往队列中。
2、另外一个线程，尝试从阻塞队列中读取数据，并设置一个超时时间。如果超时未读到数据，就继续循环等到，直到有新数据。

详细代码就不再分析了。