【一起学源码-微服务】Nexflix Eureka 源码六：在眼花缭乱的代码中，EurekaClient是如何注册的？
一枝花算不算浪漫 人气:8前言
上一讲已经讲解了EurekaClient的启动流程，到了这里已经有6篇Eureka源码分析的文章了，看了下之前的文章，感觉代码成分太多，会影响阅读，后面会只截取主要的代码，加上注释讲解。
这一讲看的是EurekaClient注册的流程，当然也是一块核心，标题为什么会写上眼花缭乱呢？关于EurekaClient注册的代码，真的不是这么容易被发现的。
如若转载 请标明来源：一枝花算不算浪漫
源码分析
如果是看过前面文章的同学，肯定会知道，Eureka Client启动流程最后是初始化DiscoveryClient这个类，那么我们就直接从这个类开始分析，后面代码都只截取重要代码，具体大家可以自行参照源码。
DiscoveryClient.java
@Inject
DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
Provider<BackupRegistry> backupRegistryProvider) {
// 省略部分代码...
this.applicationInfoManager = applicationInfoManager;
// 创建一个配置实例，这里面会有eureka的各种信息，看InstanceInfo类的注释为：The class that holds information required for registration with Eureka Server
// and to be discovered by other components.
InstanceInfo myInfo = applicationInfoManager.getInfo();
// 省略部分代码...
try {
// 支持底层的eureka client跟eureka server进行网络通信的组件
eurekaTransport = new EurekaTransport();
// 发送http请求，调用restful接口
scheduleServerEndpointTask(eurekaTransport, args);
} catch (Throwable e) {
throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
}
// 初始化调度任务
initScheduledTasks();
}上面省略了很多代码，这段代码在之前的几篇文章也都有提及，说实话看到这里 仍然一脸闷逼，入册的入口在哪呢？不急，下面慢慢分析。
DiscoveryClient.java
private void initScheduledTasks() {
// 省略大部分代码，这段代码是初始化eureka client的一些调度任务
// InstanceInfo replicator
// 创建服务拷贝副本
instanceInfoReplicator = new InstanceInfoReplicator(
this,
instanceInfo,
clientConfig.getInstanceInfoReplicationIntervalSeconds(),
2); // burstSize
// 执行线程 InitialInstanceInfoReplicationIntervalSeconds默认为40s
instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
}上面仍然是DiscoveryClient中的源码，看方法名我们知道这里肯定是初始化EurekaClient启动时的相关定时任务的。
这里主要是截取了instanceInfoReplicator初始化和执行instanceInfoReplicator.start的任务，
接着我们就可以顺着这个线先看看InstatnceInfoReplicator是何方神圣？
/**
* A task for updating and replicating the local instanceinfo to the remote server. Properties of this task are:
* - configured with a single update thread to guarantee sequential update to the remote server
* - update tasks can be scheduled on-demand via onDemandUpdate()
* - task processing is rate limited by burstSize
* - a new update task is always scheduled automatically after an earlier update task. However if an on-demand task
* is started, the scheduled automatic update task is discarded (and a new one will be scheduled after the new
* on-demand update).
*
* 用于将本地instanceinfo更新和复制到远程服务器的任务。此任务的属性是：
* -配置有单个更新线程以保证对远程服务器的顺序更新
* -可以通过onDemandUpdate（）按需调度更新任务
* -任务处理的速率受burstSize的限制
* -新更新总是在较早的更新任务之后自动计划任务。但是，如果启动了按需任务*，则计划的自动更新任务将被丢弃（并且将在新的按需更新之后安排新的任务）
*/
class InstanceInfoReplicator implements Runnable {
}这里有两个关键点：
- 此类实现了
Runnable接口，说白了就是执行一个异步线程 - 该类作用是：用于将本地instanceinfo更新和复制到远程服务器的任务
看完这两点，我又不禁陷入思考，我找的是eurekaClient注册过程，咋还跑到这个里面来了？不甘心，于是继续往下看。
InstanceInfoReplicator.start()
public void start(int initialDelayMs) {
if (started.compareAndSet(false, true)) {
instanceInfo.setIsDirty(); // for initial register
Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);
scheduledPeriodicRef.set(next);
}
}这个scheduler是一个调度任务线程池，会将this线程放入到线程池中，然后再指定时间后执行该线程的run方法。
InstanceInfoReplicator.run()
public void run() {
try {
// 刷新一下服务实例信息
discoveryClient.refreshInstanceInfo();
Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
if (dirtyTimestamp != null) {
discoveryClient.register();
instanceInfo.unsetIsDirty(dirtyTimestamp);
}
} catch (Throwable t) {
logger.warn("There was a problem with the instance info replicator", t);
} finally {
Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
scheduledPeriodicRef.set(next);
}
}看到这里是不是有种豁然开朗的感觉？看到了register就感觉到希望来了，这里使用的是DiscoveryClient.register方法，其实这里我们也可以先找DiscoveryClient中的register方法，然后再反查调用方，这也是一种好的思路呀。
DiscoveryClient.register
boolean register() throws Throwable {
logger.info(PREFIX + appPathIdentifier + ": registering service...");
EurekaHttpResponse<Void> httpResponse;
try {
// 回看eurekaTransport创建及初始化过程
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn("{} - registration failed {}", PREFIX + appPathIdentifier, e.getMessage(), e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info("{} - registration status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == 204;
}这里是使用eurekaTransport.registrationClient去进行注册，我们在最开始DiscoveryClient构造方法中已经截取了eurekaTransport创建及初始化代码，这里再贴一下：
// 支持底层的eureka client跟eureka server进行网络通信的组件
eurekaTransport = new EurekaTransport();
// 发送http请求，调用restful接口
scheduleServerEndpointTask(eurekaTransport, args);
private void scheduleServerEndpointTask(EurekaTransport eurekaTransport,
AbstractDiscoveryClientOptionalArgs args) {
// 省略大量代码
// 如果需要抓取注册表，读取其他server的注册信息
if (clientConfig.shouldRegisterWithEureka()) {
EurekaHttpClientFactory newRegistrationClientFactory = null;
EurekaHttpClient newRegistrationClient = null;
try {
newRegistrationClientFactory = EurekaHttpClients.registrationClientFactory(
eurekaTransport.bootstrapResolver,
eurekaTransport.transportClientFactory,
transportConfig
);
newRegistrationClient = newRegistrationClientFactory.newClient();
} catch (Exception e) {
logger.warn("Transport initialization failure", e);
}
// 将newRegistrationClient放入到eurekaTransport中
eurekaTransport.registrationClientFactory = newRegistrationClientFactory;
eurekaTransport.registrationClient = newRegistrationClient;
}
}到了这里，可以看到eurekaTransport.registrationClient实际就是EurekaHttpClient，不知道是我没找对地方还是什么原因，我并没有找到具体执行的实现类。
最后网上查了下，具体执行的实现类是：AbstractJersey2EurekaHttpClient
AbstractJersey2EurekaHttpClient.register()
public EurekaHttpResponse<Void> register(InstanceInfo info) {
String urlPath = "apps/" + info.getAppName();
Response response = null;
try {
// 发送请求，类似于：http://localhost:8080/v2/apps/ServiceA
// 发送的是post请求，服务实例的对象被打成了一个json发送，包括自己的主机、ip、端口号
// eureka server 就知道了这个ServiceA这个服务，有一个服务实例，比如是在192.168.31.109、host-01、8761端口
Builder resourceBuilder = jerseyClient.target(serviceUrl).path(urlPath).request();
addExtraProperties(resourceBuilder);
addExtraHeaders(resourceBuilder);
response = resourceBuilder
.accept(MediaType.APPLICATION_JSON)
.acceptEncoding("gzip")
.post(Entity.json(info));
return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
} finally {
if (logger.isDebugEnabled()) {
logger.debug("Jersey2 HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
response == null ? "N/A" : response.getStatus());
}
if (response != null) {
response.close();
}
}
}到了这里就已经真相大白了，可是 读了一通发现这个代码实在是不好理解，最后再总结一波才行。。。
总结
（1）DiscoveryClient构造函数会初始化EurekaClient相关的定时任务，定时任务里面会启动instanceInfo 互相复制的任务，就是InstanceInfoReplicator中的start()
（2）InstanceInfoReplicator的start()方法里面，将自己作为一个线程放到一个调度线程池中去了，默认是延迟40s去执行这个线程，还将isDirty设置为了ture
（3）如果执行线程的时候，是执行run()方法，线程
（3）先是找EurekaClient.refreshInstanceInfo()这个方法，里面其实是调用ApplicationInfoManager的一些方法刷新了一下服务实例的配置，看看配置有没有改变，如果改变了，就刷新一下；用健康检查器，检查了一下状态，将状态设置到了ApplicationInfoManager中去，更新服务实例的状态
（4）因为之前设置过isDirty，所以这里会执行进行服务注册
（5）服务注册的时候，是基于EurekaClient的reigster()方法去注册的，调用的是底层的TransportClient的RegistrationClient，执行了register()方法，将InstanceInfo服务实例的信息，通过http请求，调用eureka server对外暴露的一个restful接口，将InstanceInfo给发送了过去。这里找的是EurekaTransport，在构造的时候，调用了scheduleServerEndpointTask()方法，这个方法里就初始化了专门用于注册的RegistrationClient。
（6）找SessionedEurekaHttpClient调用register()方法，去进行注册，底层最终使用的AbstractJersey2EurekaHttpClient的register方法实现的
（7）eureka大量的基于jersey框架，在eureka server上提供restful接口，在eureka client如果要发送请求到eureka server的话，一定是基于jersey框架，去发送的http restful接口调用的请求
（8）真正执行注册请求的，就是eureka-client-jersey2工程里的AbstractJersey2EurekaHttpClient，请求http://localhost:8080/v2/apps/ServiceA，将服务实例的信息发送过去
eureka client这一块，在服务注册的这块代码，很多槽点：
（1）服务注册，不应该放在InstanceInfoReplicator里面，语义不明朗
（2）负责发送请求的HttpClient，类体系过于复杂，导致人根本就找不到对应的Client，最后是根据他是使用jersey框架来进行restful接口暴露和调用，才能连蒙带猜，找到真正发送服务注册请求的地方
申明
本文章首发自本人博客：https://www.cnblogs.com/wang-meng 和公众号：壹枝花算不算浪漫，如若转载请标明来源!
感兴趣的小伙伴可关注个人公众号：壹枝花算不算浪漫
加载全部内容