编程语言
首页 > 编程语言> > 【一起学源码-微服务】Nexflix Eureka 源码六:在眼花缭乱的代码中,EurekaClient是如何注册的?

【一起学源码-微服务】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 {

}

这里有两个关键点:

  1. 此类实现了Runnable接口,说白了就是执行一个异步线程
  2. 该类作用是:用于将本地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,不知道是我没找对地方还是什么原因,我并没有找到具体执行的实现类。

image.png

最后网上查了下,具体执行的实现类是: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 和公众号:壹枝花算不算浪漫,如若转载请标明来源!

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

22.jpg

标签:register,DiscoveryClient,EurekaClient,代码,Nexflix,eureka,源码,线程,eurekaTransport
来源: https://www.cnblogs.com/wang-meng/p/12111342.html