【一起学源码-微服务】Nexflix Eureka 源码六：在眼花缭乱的代码中，EurekaClient是如何注册的？

前言

上一讲已经讲解了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 和公众号：壹枝花算不算浪漫，如若转载请标明来源!

感兴趣的小伙伴可关注个人公众号：壹枝花算不算浪漫

巴特西

【一起学源码-微服务】Nexflix Eureka 源码六：在眼花缭乱的代码中，EurekaClient是如何注册的？

前言

源码分析

总结

申明

最新文章

热门文章