深度解析Eureka

向右看齐 2022-05-10 15:06 234阅读 0赞

### Eureka Register ###

##### 注册流程： #####

![在这里插入图片描述][register.png]  
客户端实例一旦创建就会向`Eureka Server`进行注册。

1.  先从`Eureka Server`拉取注册表
2.  注册之前有个`CallBack`,先执行回调
3.  客户端注册，`Eureka Server`会将客户端的注册信息保存到内存中。
4.  然后更新阀值`expectedNumberOfRenewsPerMin`和`numberOfRenewsPerMinThreshold`
5.  将新增的信息保存到`recentlyChangedQueue`队列中
6.  清空Guava缓存
7.  同步信息到集群中的其他`Eureka Server`中

--------------------

##### 源码解析 #####

DiscoveryClientApplicationReousrcePeerAwareInstaFetch Registerreturn远程post请求addInstance()register()1.保存信息到注册表2.计算保护机制阈值3.保存注册实例到队列4.清空guava缓存5.同步注册信息到集群DiscoveryClientApplicationReousrcePeerAwareInsta

对时序图中的几个类做下如下说明：  
`DiscoveryClient`是`com.netflix.discovery.EurekaClient`类的默认实现,`SpringCloud`是继承该类扩展了一些方法。提供如下能力：

*  提供获取InstanceInfo的能力
 *  提供获取有关本地客户端的数据能力（已知区域，自己的AZ等）
 *  提供注册和访问客户端的运行检查处理程序的功能

`ApplicationReousrce`是一个基于`jersey`的请求处理端，位于`com.netflix.eureka.resources`包下，该包是专门负责处理请求的。  
`PeerAwareInsta`的全称是`PeerAwareInstanceRegistryImpl`，处理服务注册，服务下线，更新，以及剔除等，实际的处理类是`com.netflix.eureka.registry.AbstractInstanceRegistry`，`PeerAwareInstanceRegistryImpl`是在该类的基础上进行扩展包装，扩展了`expectedNumberOfRenewsPerMin`和`numberOfRenewsPerMinThreshold`的更新，以及注册中心集群上的信息同步.

**1~5代表的内容：**

1.  注册表实际上`AbstractInstanceRegistry`维护的一个双层嵌套`Map`
    
        private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry
                   = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();
            ```
2.  `expectedNumberOfRenewsPerMin`和`numberOfRenewsPerMinThreshold`的重新计算，是否开启`Eureka保护机制`
3.  `recentlyChangedQueue.add()`该方法保存的是注册实例的更改，当其他`Eureka Client`与`Eureka Server`进行信息同步的时候会优先拉取`recentlyChangedQueue`队列中的内容，这个过程叫抓取增量(`delta`)
4.  `invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());`清空返回消息的缓存
5.  `PeerAwareInstanceRegistryImpl`调用 `replicateToPeers`进行集群`Eureka Server`信息的同步

在注册之前，还会存在一个预注册的处理过程，`Eureka`没有帮我们实现，这里只是提供了一个接口`com.netflix.discovery.PreRegistrationHandler`，我们可以通过实现该接口，进行一些预处理操作。

// call and execute the pre registration handler before all background tasks (inc registration) is started
    if (this.preRegistrationHandler != null) { 
    //预处理过程
       this.preRegistrationHandler.beforeRegistration();
    }
    //初始化任务，心跳检测与注册表的注册与同步
    initScheduledTasks();

public interface PreRegistrationHandler { 
        void beforeRegistration();
    }

在注册的时候，优先是考虑注册本地`区域(Region)`的`EurekaServer`的，当每个`DiscoveryClient`实例化后都会选择向`Eureka Server`进行`fetchRegistry`获取注册表。  
一般情况下，`DiscoveryClient`不会从`Eureka Server`拉取全部的注册表信息，都是拉取增量注册信息，它遵循如下的规则：

*  `delta`禁用或者`fetchRegistry`第一次调用或获取全部的注册表
 *  否则只会获取更新的增量信息，也就是`recentlyChangedQueue`维护的信息

除了向注册中心提供实例信息外，客户端还会备份一份注册表，每次实例化的时候会将实例备份一份，当所有`Eureka Server`都无法访问的时候，会从备份的注册表中获取需要调用的服务信息。

`Eureka Client`实例化后，还会启动两个定时任务，`heartbeatExecutor`与`cacheRefreshExecutor`，它们的实质是一个线程池，通过`scheduler`定时线程池来进行调用：

*  heartbeatExecutor 是进行心跳检查，负责与`Eureka Server`维持通信
 *  cacheRefreshExecutor 是负责向`Eureka Server`定期获取增量注册表,保证注册表的可靠性

scheduler = Executors.newScheduledThreadPool(2,
                        new ThreadFactoryBuilder()
                                .setNameFormat("DiscoveryClient-%d")
                                .setDaemon(true)
                                .build());
    
                heartbeatExecutor = new ThreadPoolExecutor(
                        1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                        new SynchronousQueue<Runnable>(),
                        new ThreadFactoryBuilder()
                                .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
                                .setDaemon(true)
                                .build()
                );  // use direct handoff
    
                cacheRefreshExecutor = new ThreadPoolExecutor(
                        1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                        new SynchronousQueue<Runnable>(),
                        new ThreadFactoryBuilder()
                                .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
                                .setDaemon(true)
                                .build()
                );  // use direct handoff

--------------------

### Eureka Renew ###

##### Eureka续约流程: #####

![renew.png][]

前面提到的，`Eureka Client`实例创建时，会启动一个定时任务，其中是包含着两个线程池，一个`heartbeatExecutor`,另外一个是`cacheRefreshExecutor`。`heartbeatExecutor`就是专门负责干续约这个活的。`heartbeatExecutor`会启动定时任务，频繁（`默认是30s`）的调用`renew()`与注册中心维持联系，以免`Eureka Server`的剔除任务将其剔除。

处理流程大致如下：

*  Eureka Client每隔一段时间（默认30S）就会向Eureka Server发送一个RENEW续约请求
 *  Eureka Server接收到RENEW请求后，首先找到请求中对应的实例
 *  更新实例的续约时间
 *  将更新后的实例信息同步到其他Eureka Server节点

注意：`heartbeatExecutor`与`cacheRefreshExecutor`是同时启动的，所以当第一次发送`RENEW`请求会找不到对应实例，这样它会进行注册，注册功能也是可行的。  
`heartbeatExecutor`与`cacheRefreshExecutor`分别受`registerWithEureka`与`fetchRegistry`调控。只有当其为`true`时，才会开启定时任务。

##### 源码解析 #####

DiscoveryClientInstanceResourcePeerAwareInstaLease初始化initScheduledTasks()rewen()远程put请求renewLease()registry.renew()RENEW.incrementrenewsLastMin.increment()leaseToRenew.renew()lastUpdateTimestamp更新同步注册信息到集群validateDirtyTimestamp()returnDiscoveryClientInstanceResourcePeerAwareInstaLease

--------------------

**核心代码：**  
开启续约逻辑

if (clientConfig.shouldRegisterWithEureka()) { 
                int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
                int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
                logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);
    
                // Heartbeat timer
                scheduler.schedule(
                        new TimedSupervisorTask(
                                "heartbeat",
                                scheduler,
                                heartbeatExecutor,
                                renewalIntervalInSecs,
                                TimeUnit.SECONDS,
                                expBackOffBound,
                                new HeartbeatThread()
                        ),
                        renewalIntervalInSecs, TimeUnit.SECONDS);

`HeartbeatThread`内部类是一个任务,调用`renew()`

private class HeartbeatThread implements Runnable { 
            public void run() { 
                if (renew()) { 
                    lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
                }
            }
        }

`renew()`核心逻辑，只有返回的状态码为`200`与`404`续约才能生效。

boolean renew() { 
            EurekaHttpResponse<InstanceInfo> httpResponse;
            try { 
                httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
                logger.debug("{} - Heartbeat status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
                if (httpResponse.getStatusCode() == 404) { 
                    REREGISTER_COUNTER.increment();
                    logger.info("{} - Re-registering apps/{}", PREFIX + appPathIdentifier, instanceInfo.getAppName());
                    long timestamp = instanceInfo.setIsDirtyWithTime();
                    boolean success = register();
                    if (success) { 
                        instanceInfo.unsetIsDirty(timestamp);
                    }
                    return success;
                }
                return httpResponse.getStatusCode() == 200;
            } catch (Throwable e) { 
                logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e);
                return false;
            }
        }

--------------------

### 未完待续。。。 ###

#### 参考 ####

[Eureka Register机制][Eureka Register]

[register.png]: http://static.majunwei.com/userfiles/201610292143075020/images/cms/article/2018/08/register.png
[renew.png]: http://static.majunwei.com/userfiles/201610292143075020/images/cms/article/2018/08/renew.png
[Eureka Register]: http://www.majunwei.com/view/201808130936271290.html