六、eureka客户端自动注册服务

所有文章

https://www.cnblogs.com/lay2017/p/11908715.html

正文

上一篇文章，我们稍微了解了一下eureka客户端是如何自动配置的，配置了哪些东西。在自动配置的时候会产生一个负责自动注册的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);
}

所以，我们打开EurekaAutoServiceRegistration看看

public class EurekaAutoServiceRegistration implements AutoServiceRegistration, SmartLifecycle, Ordered, SmartApplicationListener {

    private ApplicationContext context;

    private EurekaServiceRegistry serviceRegistry;

    private EurekaRegistration registration;

    public EurekaAutoServiceRegistration(ApplicationContext context, EurekaServiceRegistry serviceRegistry, EurekaRegistration registration) {
        this.context = context;
        this.serviceRegistry = serviceRegistry;
        this.registration = registration;
    }

    @Override
    public void start() {
        // ...

        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);
        }
    }

}

注册任务被委托给了serviceRegistry来做，跟进register方法

@Override
public void register(EurekaRegistration reg) {
    maybeInitializeClient(reg);

    // ...

    reg.getApplicationInfoManager().setInstanceStatus(reg.getInstanceConfig().getInitialStatus());

    reg.getHealthCheckHandler().ifAvailable(healthCheckHandler -> reg.getEurekaClient().registerHealthCheck(healthCheckHandler));
}

调用了eurekaClient原始的registerhealthCheck方法，跟进它

@Override
public void registerHealthCheck(HealthCheckHandler healthCheckHandler) {
    if (instanceInfo == null) {
        logger.error("Cannot register a healthcheck handler when instance info is null!");
    }
    if (healthCheckHandler != null) {
        // 注册心跳检查处理器
        this.healthCheckHandlerRef.set(healthCheckHandler);
        
        // schedule an onDemand update of the instanceInfo when a new healthcheck handler is registered
        if (instanceInfoReplicator != null) {
            instanceInfoReplicator.onDemandUpdate();
        }
    }
}

心跳检查处理器被设置为了成员变量，执行的核心逻辑被托付给了onDemandUpdate方法，跟进它

public boolean onDemandUpdate() {
    if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
        if (!scheduler.isShutdown()) {
            scheduler.submit(new Runnable() {
                @Override
                public void run() {
                    // ...

                    InstanceInfoReplicator.this.run();
                }
            });
            return true;
        } else {
            // ... 
        }
    } else {
        // ...
    }
}

单线程异步执行了当前类的run方法，进入run方法

public void run() {
    try {
        // 刷新实例信息
        discoveryClient.refreshInstanceInfo();

        Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
        if (dirtyTimestamp != null) {
            // 注册
            discoveryClient.register();
            instanceInfo.unsetIsDirty(dirtyTimestamp);
        }
    } catch (Throwable t) {
        // ...
    } finally {
        // 下一次延迟执行
        Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
        scheduledPeriodicRef.set(next);
    }
}

run方法每次执行都会刷新实例信息，然后调用register注册新的实例信息，最后发出下一次执行的延迟任务

跟进register方法

boolean register() throws Throwable {
    logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
    EurekaHttpResponse<Void> httpResponse;
    try {
        // 发出远程请求
        httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
    } catch (Exception e) {
        
    }
    
    return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
}

继续跟进register方法，发出了http请求，这里以jersey为例

@Override
public EurekaHttpResponse<Void> register(InstanceInfo info) {
    String urlPath = "apps/" + info.getAppName();
    ClientResponse response = null;
    try {
        Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
        addExtraHeaders(resourceBuilder);
        response = resourceBuilder
                .header("Accept-Encoding", "gzip")
                .type(MediaType.APPLICATION_JSON_TYPE)
                .accept(MediaType.APPLICATION_JSON)
                // 发出http请求
                .post(ClientResponse.class, info);
        return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
    } finally {
        if (logger.isDebugEnabled()) {
            logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
                    response == null ? "N/A" : response.getStatus());
        }
        if (response != null) {
            response.close();
        }
    }
}

http请求将进入eureka服务端，注册实例信息有兴趣可以看看eureka服务端注册服务这篇。

总结

eureka客户端自动注册服务主要是将自动配置的时候拿到的实例信息通过http请求发送给eureka服务端，默认30秒会执行一次。