服务治理 Eureka 源码解读及运行原理

75 阅读 0 评论 50 点赞

我是靠谱客的博主醉熏鸡翅，最近开发中收集的这篇文章主要介绍服务治理 Eureka 源码解读及运行原理，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

一、Eureka注册中心

Eureka是基于REST（Representational State Transfer）服务，主要以AWS云服务为支撑，提供服务发现并实现负载均衡和故障转移。我们称此服务为Eureka服务。Eureka提供了Java客户端组件，Eureka Client，方便与服务端的交互。客户端内置了基于round-robin实现的简单负载均衡。在Netflix，为Eureka提供更为复杂的负载均衡方案进行封装，以实现高可用，它包括基于流量、资源利用率以及请求返回状态的加权负载均衡。

对与Eureka的使用可参考下面我的博客：

https://blog.csdn.net/qq_43692950/article/details/107500064

二、Eureka服务注册原理

在看源码之前，先了解下Eureka Server端和Client端的运行原理：

在启动Eureka Server端的时候，会使用jax-rs 释放RESTful风格的接口供Client端调用传递信息。
当启动Eureka Client端的时候，采用 jersey 类似 HttpClient 的工具发送Http请求到服务端，并将自身的instanceId、appName、hostName、port等信息发送给Eureka Server端。
（1）当Eureka Server收到Client发送到的注册信息，会首先去一个ConcurrentHashMap类型的 registry 的容器中根据appName 取Value内容，registry的格式：key为Eureka Client的appName ，Value为Map类型的多个客户端的实例地址信息，这个Map中的key便是每个客户端 instanceId ，Value为一个Lease类型的心跳续约对象，存储着具体服务的信息，以及最后更新时间戳和最后更新时间戳等关键的信息。
（2）再收到Eureka Client请求后，先去registry 容器中取map类型的多个客户端实例信息，如果获取为空则新建一个ConcurrentHashMap 并将组建新的Lease 续约对象存入容器中，如果不为空，则再根据instanceId 获取Lease 对象更新信息。
（3）再更新完容器后，会返回给客户端204状态码，客户端收到状态码204，证明服务注册成功。

三、Eureka服务注册源码解读

在Eureka Server端启动时，采用jax-rs抛出RESTful风格接口，这个入口在com.netflix.eureka.resources.ApplicationResource 下

这里以post形式为客户端提供接口，接口中处理的逻辑现在先不看，先知道这里有个rest接口是给客户端的，下面讲到客户端发送请求了，再回来分析这边逻辑，就好理解了。
Eureka Client端启动时，主要的核心类便是com.netflix.discovery.DiscoveryClient 会触发主要的方法 DiscoveryClient类中的register()方法:

从这段代码中可以看出主要执行了eurekaTransport.registrationClient.register()这个方法传入了一个InstanceInfo 类型的对象过去，可以看下这个 InstanceInfo 对象的内容：

主要存储这当前节点的信息，其中instanceId 为节点的唯一标识，在服务端分析时会提到instanceId 。
下面在再来看到eurekaTransport.registrationClient.register()方法主要做了什么事，点进去到com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient中的register()方法中：

从上面这段代码不难看出直接使用jersey 发送Http一个请求，并将上面传过来的InstanceInfo 对象以JSON格式发送给Eureka Server端，这里的serviceUrl便是，在application.properties 或 application.yml 配置的Eureka Server端的地址，urlPath便是apps/加上服务的名称：

在这里插入图片描述
发送完请求后，如果服务端返回状态码204便代表服务注册成功。这里的204会在下面服务端讲解的时候也会说道。

当Eureka Server端收到了Client端发送的注册请求时，在上面提到会触发com.netflix.eureka.resources.ApplicationResource 的addInstance 方法，看这个方法中的代码，可以看出在上面校验一番数据后最后执行了registry.register()方法，主要传递了客户端发送来的InstanceInfo 对象，如果registry.register()执行正常，则会返回204状态码给客户端，这里正好对应上面的204状态码：

在这里插入图片描述
顺着上面方法的registry.register()方法点进去到com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl类中的register()，这里又主要调用了父类的register()方法，便来到com.netflix.eureka.registry.AbstractInstanceRegistry类中的register()方法，这个方法便是Eureka Server端注册客户端信息的主要逻辑：
在这里插入图片描述
在这个方法中，可以看到使用了读锁锁住了资源，防止并行情况下出现脏数据，然后便是到 registry() 这个容器中根据客户端的AppName 获取内容，而这个registry 容器便是ConcurrentHashMap 类型的Map容器，key为客户端的AppName，Value又为 Map类型的容器，主要是在集群情况下的存储，这个容器中的key 遍是上面提到的 instanceId ，Value为 Lease 的一个续约对象：
在这里插入图片描述
再接着上面的com.netflix.eureka.registry.AbstractInstanceRegistry类中的register()方法，如果根据AppName 获取内容为空，则创建一个新的ConcurrentHashMap 对象放入register容器中，下面又根据客户端的instanceId 获取续约对象，如果续约对象存在，则根据当前注册客户端传来InstanceInfo参数中的最后更新时间和当前续约对象中的最后更新时间对比，如果当前续约对象的最后更新时间大于了InstanceInfo中的最后更新时间，则直接将客户端传来的InstanceInfo对象覆盖：
在这里插入图片描述
如果上面的获取续约对象不存在的话，主要会更新每分钟最小续约阈值：

阈值 = 预期中每分钟发送续约消息的客户端数量 * (60s / 每分钟客户端预期续约间隔) * 续约百分比阈值。

然后便是创建续约对象，构造传入注册信息、和心跳预约时长，并将续约对象放在Map中，key为instanceId 。此时registry容器便有了客户端的地址信息
在这里插入图片描述
在registry()方法执行完，释放完read锁，如果有客户端，获取上面注册的服务，便可以取registry容器中的地址信息。

四、Eureka服务心跳续约源码解读

Eureka服务的心跳续约目的是为了保证EurekaServer端缓存的接口地址是可用的，EurekaClient默认的情况下每个30s时间会向EurekaServer端发送一个续约请求。

Eureka 续约原理

Eureka Client端启动请求Eureka Server 做服务注册时，会传递InstanceInfo 对象，在该对象中含有一个leaseInfo 的续约信息对象，其中存储着renewalIntervalInSecs 和 durationInSecs两个重要的参数，分别代表，续约周期，及预约时长，默认情况下续约周期30s，续约时长90秒。
Eureka Client端在启动后还会开启一个定时任务，周期为为默认30s 向服务端发送心跳续约信息。
Eureka Server端收到客户端心跳续约，将该地址中缓存的最后修改时间lastUpdateTimestamp改为获取当前系统时间。
Eureka Server端也会开启一个定时任务，定时清除脏地址，但前提时服务保护没有开启的情况下，默认的情况下周期为60s时间。在定时任务中，会遍历所有缓存地址，根据 获取当前系统时间 > lastUpdateTimestamp+durationInSecs*1000，也就是当前系统时间是否大于该服务最后更新时间加上续约时长，如果大于则认为该服务地址为脏地址，然后将该脏地址放入到一个新的集合中，脏地址放入到一个新的集合中。

Eureka 续约源码解读

在Eureka Client端的com.netflix.discovery.DiscoveryClient类中有个initScheduledTasks()方法

在Eureka Client端定时任务是采用的线程池的定时任务来实现的，定时任务的周期默认为30s，

在从源码中看，可以看到其实是创建了两个定时任务，第一个为刷新缓存的定时任务，续约的定时任务为第二个，也就是上面截图的代码，从上面应该能看到创建定时任务，执行业务逻辑应该就在在HeartbeatThread 这个类中，点击去可以发现，又调用了renew()方法，并且如果该方法执行成功返回true，则直接将当前时间付给lastSuccessfulHeartbeatTimestamp，代表续约成功时间戳。

接着点进去renew()这个方法：

可以看到其实是执行了一次Http请求，可以点到eurekaTransport.registrationClient.sendHeartBeat这个方法到com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient类下的sendHeartBeat下：

其实就是通过jersey向服务端发送了一个Http请求。
Eureka Server端收到Client端的续约请求后，会走com.netflix.eureka.registry.AbstractInstanceRegistry类中的renew()方法：

根据代码的逻辑来，最后执行了leaseToRenew.renew();这个方法，可以点进去看这个方法的逻辑，就是更新了该续命服务的最后更新时间来延长寿命。

五、Eureka Server端服务剔除

上面提到在Eureka Server 端也会开启一个定时任务，定期清除脏地址，这个定时任务在com.netflix.eureka.registry.AbstractInstanceRegistry的postInit()方法中：
在这里插入图片描述

在这里插入图片描述
这里的定时任务，直接采用的java 的Timer来实现的，相对比客户端采用的线程池的还是有一定的不一样，上面定时任务，明显执行了EvictionTask中的事件，点进去在run()方法中又调用了evict() 方法，清理过期地址的主要逻辑便在该方法中：
在这里插入图片描述

在evict()主要的东西便是上面截图的代码，在这边会遍历整个registry容器，获取到每个具体的服务的Lease对象，在判断是否为过期脏地址是执行了lease.isExpired()方法，先看下里面有什么内容：

逻辑就是：系统当前时间 > 最后更新时间 + 过期时间 + 预留时间（集群同步预留），如果过期之后存入到过期列表集合中，实际本没有剔除该地址，因为在实际剔除地址之前，判断是否开启了eureka自我保护机制，如果开启了自我保护机制则不会剔除该脏地址。如果没有关闭自我保护机制，则会采用随机算法剔除脏地址，而不是全部剔除脏地址。

这个方法再往下看，便是随机算法剔除脏地址的逻辑：
在这里插入图片描述
上面有几个变量需要关注下，registrySizeThreshold为根据阈值计算可以被剔除的服务数量最大值，它等于注册服务的数量 * 续约阈值（默认0.85），evictionLimit剔除后剩余最小数量，expiredLeases.size() 剔除列表的数量，知道这几个参数的含义，应该不难看懂下面循环中的逻辑了，随机计算出要删除的下标，对换位置，然后取出该服务在registry的外层Map容器 key AppName和内层Map容器key instanceId，直接在internalCancel()方法中将该服务地址remove掉：
在这里插入图片描述

六、Eureka 的服务下线

EurekaClient停止的时候，不会立即停止，而是会先给EurekaServer端发送一个通知剔除该地址，也在com.netflix.discovery.DiscoveryClient 的 unregister()方法中：
在这里插入图片描述
和上面一样发送了一个Http请求给服务端，当服务端收到请求，便会触发服务端的 com.netflix.eureka.registry.AbstractInstanceRegistry 的cancel()方法：

在cancel方法中调用了又internalCancel方法，删除服务。

七、Eureka集群环境原理

Eureka集群采用相互注册的原理，每个节点都是均等方式，从而实现保证AP模式。

当EurekaClient实现服务注册时候，会将该接口注册到Eureka Server端
Eureka Server端会将该地址缓存到Map中，同时遍历所有副本节点地址
发送rest请求到副本节点实现数据同步

主要逻辑在com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl类下的register()方法，在该方法中在父类的register()方法中完成服务的注册后，执行了replicateToPeers()来实现集群的同步：
在这里插入图片描述

在replicateToPeers()方法中主要是上面我圈起来的for循环，循环每个副本节点，发送Http请求进行同步信息，一步步点进去可以看下下面代码：
在这里插入图片描述
此时便回到了最初的客户端注册时的场景，在集群情况下，数据同步其实就和客户端注册一样，只不过此时的客户端时集群中的一个节点。