Eureka 系列(07)服务注册与主动下线

Posted binarylei

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Eureka 系列(07)服务注册与主动下线相关的知识,希望对你有一定的参考价值。

Eureka 系列(07)服务注册与主动下线

[TOC]

Spring Cloud 系列目录 - Eureka 篇

在上一篇 Eureka 系列(05)消息广播 中对 Eureka 消息广播的源码进行了分析,之后的几篇文章会具体分析本地服务注册、主动下线、心跳续约、自动过期等的实现机制。

  • PeerAwareInstanceRegistryImpl 负责集群内部消息通信。
  • AbstractInstanceRegistry 负责本地服务信息管理,这也是之后几篇文章关注的重点。
表1:Eureka OPEN API
资源 功能 url
ApplicationsResource 获取全部或增量服务实例信息 GET /apps
GET /apps/delta
ApplicationResource 1. 获取单个应用的信息
2. 注册实例信息
GET /apps/appName
POST /apps/appName
InstanceResource 服务实例的CURD:
1. 获取实例的信息
2.修改服务实例元信息
3. 删除实例信息,服务下线
4. 发送心跳
GET /apps/appName/id
PUT /apps/appName/id/metadata
DELETE /apps/appName/id
PUT /apps/appName/id
InstancesResource 直接根据实例id获取实例信息 GET /instances/id
PeerReplicationResource 集群内部批量数据同步 POST /peerreplication/batch
ServerInfoResource ??? POST /serverinfo/statusoverrides
StatusResource ??? GET /statusoverrides

注: 表示应用名称或服务id, 表示实例id。eg: http://localhost:8080/eureka/apps

1. 服务注册

1.1 服务实例注册流程

图1:Eureka 服务实例注册时序图
sequenceDiagram participant ApplicationResource participant PeerAwareInstanceRegistryImpl participant AbstractInstanceRegistry participant PeerEurekaNode note over ApplicationResource: POST:/euraka/apps/appName<br/>addInstance(instanceInfo,isReplication) ApplicationResource ->> PeerAwareInstanceRegistryImpl: 注册请求:register(instanceInfo,isReplication) PeerAwareInstanceRegistryImpl ->> AbstractInstanceRegistry: 1. 本地数据更新: register(instanceInfo,leaseDuration,isReplication) loop 同步到其它 Eureka Server 节点 PeerAwareInstanceRegistryImpl ->> PeerAwareInstanceRegistryImpl: 2.1 数据同步:replicateInstanceActionsToPeers PeerAwareInstanceRegistryImpl ->> PeerEurekaNode: 2.2 register(instanceInfo) -> POST:/euraka/apps/appName end

总结: Eureka Web 使用的是 Jersey 容器,服务注册的请求入口是 ApplicationResource 的 register 方法,请求的路径是 POST:/euraka/apps/appName

1.2 ApplicationResource

// ApplicationResource HTTP请求入口
@POST
@Consumes("application/json", "application/xml")
public Response addInstance(InstanceInfo info,
	@HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) 
    ... // 数据检验
    registry.register(info, "true".equals(isReplication));
    return Response.status(204).build();  // 204 to be backwards compatible

总结: ApplicationResource 入口主要是进行参数检验,主要的逻辑都委托给了 PeerAwareInstanceRegistryImpl 完成。

注意:isReplication 参数,如果是客户端的请求则为 false,表示需要将这个消息广播给其它服务器。如果是集群内部消息广播则为true,表示不再需要继续广播,否则会造成循环广播的问题。

// PeerAwareInstanceRegistryImpl 默认注册器实现
@Override
public void register(final InstanceInfo info, final boolean isReplication) 
    // 租约的过期时间,默认90秒
    int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
    if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) 
        // 如果客户端自定义了,那么以客户端为准
        leaseDuration = info.getLeaseInfo().getDurationInSecs();
    
    // 本地注册
    super.register(info, leaseDuration, isReplication);
    // 消息广播
    replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);

总结: PeerAwareInstanceRegistryImpl 这个类应该都很熟悉了,主要是负责进行集群间内部通信的,其父类 AbstractInstanceRegistry 则负责本地服务信息管理,也是本文的研究重点。

1.3 AbstractInstanceRegistry

在 Eureka 中,服务注册信息存储在内存中,数据结构为 ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry,Map 嵌套了两层,外层是的 Key 是 appName,内存的 Key 是 InstanceId。

private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry
            = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) 
    try 
        read.lock();
        // 1. 获取该服务对应的所有服务实例,如果不存在就创建一个新的Map
        Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
        REGISTER.increment(isReplication);
        if (gMap == null) 
            final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
            gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
            if (gMap == null) 
                gMap = gNewMap;
            
        
        
        // 2. 两种情况:一是实例已经注册,二是实例没有注册
        Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
        // 2.1 实例已经注册,就需要PK,PK原则:谁最后一次更新就是谁赢
        //     也就是说如果已经注册的实例最近更新了,就不用重新更新了
        if (existingLease != null && (existingLease.getHolder() != null)) 
            Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
            Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
            // 已经注册的实例PK赢了
            if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) 
                registrant = existingLease.getHolder();
            
        // 2.2 没有注册,很好处理。更新注册的实例个数
         else 
            synchronized (lock) 
                if (this.expectedNumberOfClientsSendingRenews > 0) 
                    this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
                    updateRenewsPerMinThreshold();
                
            
        
        
        // 3. 更新注册信息(核心步骤)
        Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
        if (existingLease != null) 
            lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
        
        // (核心步骤)
        gMap.put(registrant.getId(), lease);
        // 添加到最近的注册队列里面去,以时间戳作为Key,名称作为value,主要是为了运维界面的统计数据
        synchronized (recentRegisteredQueue) 
            recentRegisteredQueue.add(new Pair<Long, String>(
                System.currentTimeMillis(),
                registrant.getAppName() + "(" + registrant.getId() + ")"));
        
        
        // 4. 更新实例状态 InstanceStatus
        if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) 
            if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) 
                overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
            
        
        InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
        if (overriddenStatusFromMap != null) 
            registrant.setOverriddenStatus(overriddenStatusFromMap);
        

        InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
        registrant.setStatusWithoutDirty(overriddenInstanceStatus);

        if (InstanceStatus.UP.equals(registrant.getStatus())) 
            lease.serviceUp();
        
        
        // 5. 清理缓存等善后工作
        registrant.setActionType(ActionType.ADDED);
        // 租约变更记录队列,记录了实例的每次变化, 用于注册信息的增量获取
        recentlyChangedQueue.add(new RecentlyChangedItem(lease));
        registrant.setLastUpdatedTimestamp();
        // 清理缓存 ,传入的参数为key
        invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
     finally 
        read.unlock();
    

总结: 前三步的逻辑都很清楚,目的就是更新内存中的实例信息,只是需要注意实例已经存在的情况下需要 PK 一下,原则就是谁最后一次更新就是谁赢。

第四步更新服务实例的状态,OverriddenStatus 参考 InstanceInfo 中 OverriddenStatus 的作用

每五步清理缓存等善后工作。目前来说,到 gMap.put(registrant.getId(), lease)这一步就够了。

2. 主动下线

服务下线对应的 OPEN API 为 DELETE /apps/appName/id

protected boolean internalCancel(String appName, String id, boolean isReplication) 
    try 
        read.lock();
        CANCEL.increment(isReplication);
        // 1. 清空registry中注册的实例信息(核心步骤)
        Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
        Lease<InstanceInfo> leaseToCancel = null;
        if (gMap != null) 
            leaseToCancel = gMap.remove(id);
        
        // 2. 添加到 recentCanceledQueue 队列中
        synchronized (recentCanceledQueue) 
            recentCanceledQueue.add(new Pair<Long, String>(System.currentTimeMillis(), appName + "(" + id + ")"));
        
        InstanceStatus instanceStatus = overriddenInstanceStatusMap.remove(id);
        if (leaseToCancel == null) 
            CANCEL_NOT_FOUND.increment(isReplication);
            return false;
         else 
            // 3. 和注册时一样,也要做一下清除缓存等善后工作
            leaseToCancel.cancel();
            InstanceInfo instanceInfo = leaseToCancel.getHolder();
            String vip = null;
            String svip = null;
            if (instanceInfo != null) 
                instanceInfo.setActionType(ActionType.DELETED);
                recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel));
                instanceInfo.setLastUpdatedTimestamp();
                vip = instanceInfo.getVIPAddress();
                svip = instanceInfo.getSecureVipAddress();
            
            invalidateCache(appName, vip, svip);
            return true;
        
     finally 
        read.unlock();
    

总结: 服务的注册和主动下线的逻辑还是很清楚的。**目前来说,到 gMap.remove(id)这一步就够了。**至于细节以后真正用到 Eureka 再继续深入研究。


每天用心记录一点点。内容也许不重要,但习惯很重要!

以上是关于Eureka 系列(07)服务注册与主动下线的主要内容,如果未能解决你的问题,请参考以下文章

Eureka服务下线源码解析

Eureka总结笔记

springboot-服务治理

EUREKA 删除 or 强制下线/上线 实例

服务注册与发现组件 Eureka 客户端实现原理解析

Eureka剔除和恢复服务