Spring Cloud学习--Spring Cloud Eureka服务注册

╰半橙微兮° 2022-05-26 10:10 168阅读 0赞

这篇博客我们通过分析一下Spring Cloud Eureka相关的源码来了解一下Spring Cloud注册中心是如何工作的及客户端是如何与注册中心进行交互的。

### 服务注册 ###

在服务治理与发现中，服务提供者需要一个注册中心将其一些服务信息（服务名，IP，端口号，版本号等信息）注册到中超中心中，比如Dubbo中常见的会将接口等详细信息注册到ZK中，接下来我们通过分析eureka client的实现机制了解一下eureka client是如何将自己的信息注册到注册中心及从注册中心中获取到想要的服务信息的。

1、Eureka Client服务注册

在eureka client在启动的时候会进行初始化DiscoveryClient，在进行DiscoverClient初始化的时候会初始化定时任务线程池并启动三个线程

（1）CacheRefreshThread：在定时线程池中执行，定时刷新本地缓存中的数据（会有一篇专门的博客介绍）

（2）HeartbeatThread：在定时线程池中执行，定时维持客户端与注册中心的心跳保活（会有一篇专门的博客介绍）

（3）InstanceInfoReplicator：此线程自己维持了一个定时任务，会定时检查注册中心的客户端信息是否为脏数据的，并进行数据注册操作。

@Inject
    DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                        Provider<BackupRegistry> backupRegistryProvider) {
           
    
    		//省略部分代码，在DiscoveryClient初始化时会创建几个线程池，用于定时刷新本地缓存数据及与注册中心进行心跳交互
            initScheduledTasks();
    
          
        }

private void initScheduledTasks() {
            if (clientConfig.shouldFetchRegistry()) {
                // registry cache refresh timer
                int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
                int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
    			//创建定时任务，定时刷新本地数据
                scheduler.schedule(
                        new TimedSupervisorTask(
                                "cacheRefresh",
                                scheduler,
                                cacheRefreshExecutor,
                                registryFetchIntervalSeconds,
                                TimeUnit.SECONDS,
                                expBackOffBound,
                                new CacheRefreshThread()
                        ),
                        registryFetchIntervalSeconds, TimeUnit.SECONDS);
            }
    
            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);
    
                // InstanceInfo replicator
    			//创建服务注册线程
                instanceInfoReplicator = new InstanceInfoReplicator(
                        this,
                        instanceInfo,
                        clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                        2); // burstSize
    
                statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
                    @Override
                    public String getId() {
                        return "statusChangeListener";
                    }
    
                    @Override
                    public void notify(StatusChangeEvent statusChangeEvent) {
                        if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
                                InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
                            // log at warn level if DOWN was involved
                            logger.warn("Saw local status change event {}", statusChangeEvent);
                        } else {
                            logger.info("Saw local status change event {}", statusChangeEvent);
                        }
                        instanceInfoReplicator.onDemandUpdate();
                    }
                };
    
                if (clientConfig.shouldOnDemandUpdateStatusChange()) {
                    applicationInfoManager.registerStatusChangeListener(statusChangeListener);
                }
    			//开启服务注册线程，执行服务注册
                instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
            } else {
                logger.info("Not registering with Eureka server per configuration");
            }
        }

接下来我们看看在InstanceInfoReplicator中做了什么处理操作。其实InstanceInfoReplicator中的操作也是很简单的，在调用线程的start方法时会创建一个定时任务线程池，定时去执行此线程，在run方法中会判断是否存在脏时间戳，如果存在则调用DiscoveryClient的register方法进行服务注册。

class InstanceInfoReplicator implements Runnable {
        private static final Logger logger = LoggerFactory.getLogger(InstanceInfoReplicator.class);
    
        private final DiscoveryClient discoveryClient;
        private final InstanceInfo instanceInfo;
    
        private final int replicationIntervalSeconds;
        private final ScheduledExecutorService scheduler;
        private final AtomicReference<Future> scheduledPeriodicRef;
    
        private final AtomicBoolean started;
        private final RateLimiter rateLimiter;
        private final int burstSize;
        private final int allowedRatePerMinute;
    
        InstanceInfoReplicator(DiscoveryClient discoveryClient, InstanceInfo instanceInfo, int replicationIntervalSeconds, int burstSize) {
            this.discoveryClient = discoveryClient;
            this.instanceInfo = instanceInfo;
            this.scheduler = Executors.newScheduledThreadPool(1,
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-InstanceInfoReplicator-%d")
                            .setDaemon(true)
                            .build());
    
            this.scheduledPeriodicRef = new AtomicReference<Future>();
    
            this.started = new AtomicBoolean(false);
            this.rateLimiter = new RateLimiter(TimeUnit.MINUTES);
            this.replicationIntervalSeconds = replicationIntervalSeconds;
            this.burstSize = burstSize;
    
            this.allowedRatePerMinute = 60 * this.burstSize / this.replicationIntervalSeconds;
            logger.info("InstanceInfoReplicator onDemand update allowed rate per min is {}", allowedRatePerMinute);
        }
    	//调用线程的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);
            }
        }
    
        public void stop() {
            scheduler.shutdownNow();
            started.set(false);
        }
    
        public boolean onDemandUpdate() {
            if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
                if (!scheduler.isShutdown()) {
                    scheduler.submit(new Runnable() {
                        @Override
                        public void run() {
                            logger.debug("Executing on-demand update of local InstanceInfo");
        
                            Future latestPeriodic = scheduledPeriodicRef.get();
                            if (latestPeriodic != null && !latestPeriodic.isDone()) {
                                logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
                                latestPeriodic.cancel(false);
                            }
        
                            InstanceInfoReplicator.this.run();
                        }
                    });
                    return true;
                } else {
                    logger.warn("Ignoring onDemand update due to stopped scheduler");
                    return false;
                }
            } else {
                logger.warn("Ignoring onDemand update due to rate limiter");
                return false;
            }
        }
    
    	//线程执行时会调用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);
            }
        }
    
    }

服务注册客户端最重要的实现逻辑在DiscoveryClient的register方法，接下来我们看看在register中做了什么处理操作。在register方法中处理的方法很简单，就是将服务实例信息instanceInfo通过http请求发送到注册中心，如果返回的状态码是204则表示注册成功，否则表示失败。

//在register方法中处理的方法很简单，就是将服务实例信息instanceInfo通过http请求发送到注册中心，如果返回的状态码是204则表示注册成功，否则表示失败
     boolean register() throws Throwable {
            logger.info(PREFIX + appPathIdentifier + ": registering service...");
            EurekaHttpResponse<Void> httpResponse;
            try {
    			//将服务实例信息instanceInfo通过httpClient注册到注册中心
                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());
            }
    		//判断状态码是否是204，如果是则表示成功
            return httpResponse.getStatusCode() == 204;
        }

向注册中心的地址如下图：http://localhost:1001/eureka/apps/COMPUTER-SERVICE，http方法为POST，消息体为实例信息instanceInfo，通过将实例数据信息通过POST提交到注册中心来实现服务注册的机制。

![70][]

2、Eureka Server服务注册

Eureka Server通过对外提供POST请求来接收客户端的服务注册信息完成服务注册操作，接下来我们看看Eureka Server在服务注册方面做了哪些处理操作。

Eureka Server对外暴露POST服务接收服务注册的实现类是ApplicationResource，在ApplicationResource中提供了addInstance方法接收服务注册操作。

@POST
        @Consumes({"application/json", "application/xml"})
        public Response addInstance(InstanceInfo info,
                                    @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
            logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
            // validate that the instanceinfo contains all the necessary required fields
            if (isBlank(info.getId())) {
                return Response.status(400).entity("Missing instanceId").build();
            } else if (isBlank(info.getHostName())) {
                return Response.status(400).entity("Missing hostname").build();
            } else if (isBlank(info.getIPAddr())) {
                return Response.status(400).entity("Missing ip address").build();
            } else if (isBlank(info.getAppName())) {
                return Response.status(400).entity("Missing appName").build();
            } else if (!appName.equals(info.getAppName())) {
                return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();
            } else if (info.getDataCenterInfo() == null) {
                return Response.status(400).entity("Missing dataCenterInfo").build();
            } else if (info.getDataCenterInfo().getName() == null) {
                return Response.status(400).entity("Missing dataCenterInfo Name").build();
            }
    
            // handle cases where clients may be registering with bad DataCenterInfo with missing data
            DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
            if (dataCenterInfo instanceof UniqueIdentifier) {
                String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
                if (isBlank(dataCenterInfoId)) {
                    boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
                    if (experimental) {
                        String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
                        return Response.status(400).entity(entity).build();
                    } else if (dataCenterInfo instanceof AmazonInfo) {
                        AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
                        String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);
                        if (effectiveId == null) {
                            amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());
                        }
                    } else {
                        logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
                    }
                }
            }
    		//保存服务提供者的信息
            registry.register(info, "true".equals(isReplication));
            return Response.status(204).build();  // 204 to be backwards compatible
        }

调用InstanceRegistry的register完成服务的注册工作。

@Override
    	public void register(final InstanceInfo info, final boolean isReplication) {
    		//通知服务注册，利用spring的Event进行事件通知
    		handleRegistration(info, resolveInstanceLeaseDuration(info), isReplication);
    		//调用父类的register方法
    		super.register(info, isReplication);
    	}

在PeerAwareInstanceRegistryImpl中调用register方法，在register中首先调用服务的register方法进行服务注册，然后将注册的服务通知到注册中心的所有节点。

@Override
        public void register(final InstanceInfo info, final boolean isReplication) {
            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);
        }

在replicateToPeers中会将注册的节点信息通知到注册中心的所有节点

private void replicateToPeers(Action action, String appName, String id,
                                      InstanceInfo info /* optional */,
                                      InstanceStatus newStatus /* optional */, boolean isReplication) {
            Stopwatch tracer = action.getTimer().start();
            try {
                if (isReplication) {
                    numberOfReplicationsLastMin.increment();
                }
                // If it is a replication already, do not replicate again as this will create a poison replication
                if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {
                    return;
                }
    			//依次通知所有的节点
                for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {
                    // If the url represents this host, do not replicate to yourself.
                    if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {
                        continue;
                    }
                    replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
                }
            } finally {
                tracer.stop();
            }
        }
    
    //根据action动作类型类型来依次通知
     private void replicateInstanceActionsToPeers(Action action, String appName,
                                                     String id, InstanceInfo info, InstanceStatus newStatus,
                                                     PeerEurekaNode node) {
            try {
                InstanceInfo infoFromRegistry = null;
                CurrentRequestVersion.set(Version.V2);
                switch (action) {
                    case Cancel:
                        node.cancel(appName, id);
                        break;
                    case Heartbeat:
                        InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);
                        infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                        node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
                        break;
                    case Register:
                        node.register(info);
                        break;
                    case StatusUpdate:
                        infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                        node.statusUpdate(appName, id, newStatus, infoFromRegistry);
                        break;
                    case DeleteStatusOverride:
                        infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                        node.deleteStatusOverride(appName, id, infoFromRegistry);
                        break;
                }
            } catch (Throwable t) {
                logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
            }
        }

在父类AbstractInstanceRegistry中会调用register方法将服务注册者的信息添加到ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry中，完成服务注册的操作。

public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
            try {
    			//在注册信息保存时添加锁
                read.lock();
    			//首先查看是否存在该服务
                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>>();
    				//将注册信息添加到map中
                    gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
                    if (gMap == null) {
                        gMap = gNewMap;
                    }
                }
                Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
                // Retain the last dirty timestamp without overwriting it, if there is already a lease
                if (existingLease != null && (existingLease.getHolder() != null)) {
                    Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
                    Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
                    logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
    
                    // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
                    // InstanceInfo instead of the server local copy.
                    if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
                        logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
                                " than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
                        logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
                        registrant = existingLease.getHolder();
                    }
                } else {
                    // The lease does not exist and hence it is a new registration
                    synchronized (lock) {
                        if (this.expectedNumberOfRenewsPerMin > 0) {
                            // Since the client wants to cancel it, reduce the threshold
                            // (1
                            // for 30 seconds, 2 for a minute)
                            this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin + 2;
                            this.numberOfRenewsPerMinThreshold =
                                    (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
                        }
                    }
                    logger.debug("No previous lease information found; it is new registration");
                }
                Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
                if (existingLease != null) {
                    lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
                }
    			//将注册信息添加到map中
                gMap.put(registrant.getId(), lease);
                synchronized (recentRegisteredQueue) {
                    recentRegisteredQueue.add(new Pair<Long, String>(
                            System.currentTimeMillis(),
                            registrant.getAppName() + "(" + registrant.getId() + ")"));
                }
                // This is where the initial state transfer of overridden status happens
                if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
                    logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
                                    + "overrides", registrant.getOverriddenStatus(), registrant.getId());
                    if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
                        logger.info("Not found overridden id {} and hence adding it", registrant.getId());
                        overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
                    }
                }
                InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
                if (overriddenStatusFromMap != null) {
                    logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
                    registrant.setOverriddenStatus(overriddenStatusFromMap);
                }
    
                // Set the status based on the overridden status rules
                InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
                registrant.setStatusWithoutDirty(overriddenInstanceStatus);
    
                // If the lease is registered with UP status, set lease service up timestamp
                if (InstanceStatus.UP.equals(registrant.getStatus())) {
                    lease.serviceUp();
                }
                registrant.setActionType(ActionType.ADDED);
                recentlyChangedQueue.add(new RecentlyChangedItem(lease));
                registrant.setLastUpdatedTimestamp();
                invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
                logger.info("Registered instance {}/{} with status {} (replication={})",
                        registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
            } finally {
                read.unlock();
            }
        }

总结：Eureka 提供的注册机制还是挺简单的，最终注册信息都会在Map中保存，做到了注册中心的轻量级，现在来看很多处理机制需要注册者去做处理操作的。

（盗图）

![70 1][]

[70]: /images/20220526/f63088f32fb841eb85e510a00bc1b186.png
[70 1]: /images/20220526/44686eabe88f42efb2fd7dc00dc8d048.png