Eureka可用性源码分析

zhaozhen

发布于 2021-07-15 10:46:32

3310

发布于 2021-07-15 10:46:32

元数据

除了普通的基础设置之外，eureka支持自定义元数据。配置方式如下

eureka:
 instance:
    metadata-map:
       cluster: cl1
       name: zhaozhen

获取元数据代码

        List<ServiceInstance> list = discoveryClient.getInstances("zhao-service-resume");
        ServiceInstance serviceInstance = list.get(0);
        list.stream().forEach(s->{
            System.out.println(s.getMetadata());
        });

在调用时通过断点可以知道具体的元数据。在实际使用过程中，我们可以针对配置的不同元数据采取不同的执行

可用性

从技术网站上搜到的一个面试题就有这样的问题:eureka怎么保证可用性. 众所周知，eureka采用的是AP模式，实现高可用最好的方式就是利用最少三台eureke server实例，实现两两之间的服务注册。从而达到同步数据的目的那么这就涉及到如下的方面

eureka client和eureka server之间如何进行通信
eureka注册在客户端和服务端分别怎么操作实现可用性的
eureka续约/心跳在客户端和服务端分别怎么操作实现可用性的
eureka下线是怎么操作的

eureka client和eureka server之间如何进行通信

通过查询各种资料并追踪自动配置类发现，eureka和eureka之间的通信是采用类似springmvc的Jersey框架暴露接口进行通信的。通信的形式基本类似于我们使用http进行请求的方式。在EurekaServerAutoConfiguration中通过注入FilterRegistrationBean实现了在filter中加入包含了指定包名下的所有的Jersey的外部接口

/**
  * Register the Jersey filter
  */
 @Bean
 public FilterRegistrationBean jerseyFilterRegistration(
   javax.ws.rs.core.Application eurekaJerseyApp) {
  FilterRegistrationBean bean = new FilterRegistrationBean();
  bean.setFilter(new ServletContainer(eurekaJerseyApp));
  bean.setOrder(Ordered.LOWEST_PRECEDENCE);
  bean.setUrlPatterns(
    Collections.singletonList(EurekaConstants.DEFAULT_PREFIX + "/*"));

  return bean;
 }

 /**
  * Construct a Jersey {@link javax.ws.rs.core.Application} with all the resources
  * required by the Eureka server.
  */
 @Bean
 public javax.ws.rs.core.Application jerseyApplication(Environment environment,
   ResourceLoader resourceLoader) {

  ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(
    false, environment);

  // Filter to include only classes that have a particular annotation.
  //
  provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));
  provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));

  // Find classes in Eureka packages (or subpackages)
  //
  Set<Class<?>> classes = new HashSet<>();
  for (String basePackage : EUREKA_PACKAGES) {
   Set<BeanDefinition> beans = provider.findCandidateComponents(basePackage);
   for (BeanDefinition bd : beans) {
    Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(),
      resourceLoader.getClassLoader());
    classes.add(cls);
   }
  }

  // Construct the Jersey ResourceConfig
  //
  Map<String, Object> propsAndFeatures = new HashMap<>();
  propsAndFeatures.put(
    // Skip static content used by the webapp
    ServletContainer.PROPERTY_WEB_PAGE_CONTENT_REGEX,
    EurekaConstants.DEFAULT_PREFIX + "/(fonts|images|css|js)/.*");

  DefaultResourceConfig rc = new DefaultResourceConfig(classes);
  rc.setPropertiesAndFeatures(propsAndFeatures);

  return rc;
 }

代码中扫描的EUREKA_PACKAGES（private static final String[] EUREKA_PACKAGES = new String[] { "com.netflix.discovery", "com.netflix.eureka" };）即是Jersey框架的具体的接口类

另外可以提一点的就是，eureka对外暴露的dashboard依然采用的是springmvc的controller形式。具体的可以看到在EurekaServerAutoConfiguration中注入的EurekaController

 @Bean
 @ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled", matchIfMissing = true)
 public EurekaController eurekaController() {
  return new EurekaController(this.applicationInfoManager);
 }

感兴趣的可以再研究下后续EurekaController的内部实现

eureka注册在客户端和服务端分别怎么操作实现可用性的

服务每隔30秒会向注册中⼼续约(⼼跳)⼀次（也称为报活），如果没有续约，租约在90秒后到期，然后服务会被失效。每隔30秒的续约操作我们称之为⼼跳检测首先在服务端，通过上述的Jersey框架暴露的接口进行注册，在ApplicationResource中通过addInstance进行注册，在这个过程中另一个eureka server也相当于是一个eureka client，同样会进行注册

通过addInstance中的register方法，一直向下调试到PeerAwareInstanceRegistryImpl的replicateInstanceActionsToPeers相互注册方法

   /**
     * Replicates all instance changes to peer eureka nodes except for
     * replication traffic to this node.
     *
     */
    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);
        }
    }

此时，注册时，进入的是Register

    public void register(final InstanceInfo info) throws Exception {
        long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
        batchingDispatcher.process(
                taskId("register", info),
                new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {
                    public EurekaHttpResponse<Void> execute() {
                        return replicationClient.register(info);
                    }
                },
                expiryTime
        );
    }

查阅源码可知此处的getLeaseRenewalOf(info)的默认值为90秒,这就印证了90秒到期的说法

    private static int getLeaseRenewalOf(InstanceInfo info) {
        return (info.getLeaseInfo() == null ? Lease.DEFAULT_DURATION_IN_SECS : info.getLeaseInfo().getRenewalIntervalInSecs()) * 1000;
    }

发起请求

    @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)
                    .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();
            }
        }
    }

发起请求的地址可以追踪到的是ApplicationsResource中的

    @Path("{appId}")
    public ApplicationResource getApplicationResource(
            @PathParam("version") String version,
            @PathParam("appId") String appId) {
        CurrentRequestVersion.set(Version.toEnum(version));
        return new ApplicationResource(appId, serverConfig, registry);
    }

此处重新构建了一个ApplicationResource对象。并将服务的信息配置等传递到application中，等待后续使用分析完这一段之后,我对addInstance如何接收请求的还是有疑问，经过断点调试发现，这个过程实际上是通过EurekaServerAutoConfiguration引入的 EurekaServerInitializerConfiguration来完成的，

@Configuration
public class EurekaServerInitializerConfiguration
  implements ServletContextAware, SmartLifecycle, Ordered {
}

EurekaServerInitializerConfiguration实现了SmartLifecycle方法，start方法会再容器初始化时执行。而start方法的内容

@Override
 public void start() {
  new Thread(new Runnable() {
   @Override
   public void run() {
    try {
     //TODO: is this class even needed now?
     eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
     log.info("Started Eureka Server");

     publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
     EurekaServerInitializerConfiguration.this.running = true;
     publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
    }
    catch (Exception ex) {
     // Help!
     log.error("Could not initialize Eureka servlet context", ex);
    }
   }
  }).start();
 }

具体的业务内容在

 public void contextInitialized(ServletContext context) {
  try {
   initEurekaEnvironment();
   initEurekaServerContext();

   context.setAttribute(EurekaServerContext.class.getName(), this.serverContext);
  }
  catch (Throwable e) {
   log.error("Cannot bootstrap eureka server :", e);
   throw new RuntimeException("Cannot bootstrap eureka server :", e);
  }
 }

第一步initEurekaEnvironment为初始化环境，第二步initEurekaServerContext为业务操作而随后的操作中最主要的是

  int registryCount = this.registry.syncUp();
  this.registry.openForTraffic(this.applicationInfoManager, registryCount);

  // Register all monitoring statistics.
  EurekaMonitors.registerAllStats();

openForTraffic中主要是为开启服务通信做准备

  @Override
    public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
        // Renewals happen every 30 seconds and for a minute it should be a factor of 2.
        this.expectedNumberOfClientsSendingRenews = count;
        updateRenewsPerMinThreshold();
        logger.info("Got {} instances from neighboring DS node", count);
        logger.info("Renew threshold is: {}", numberOfRenewsPerMinThreshold);
        this.startupTime = System.currentTimeMillis();
        if (count > 0) {
            this.peerInstancesTransferEmptyOnStartup = false;
        }
        DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
        boolean isAws = Name.Amazon == selfName;
        if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
            logger.info("Priming AWS connections for all replicas..");
            primeAwsReplicas(applicationInfoManager);
        }
        logger.info("Changing status to UP");
        applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
        super.postInit();
    }

引发向addIntsance发起请求的就是 applicationInfoManager.setInstanceStatus(InstanceStatus.UP);这个方法内部执行一串事件其中就有向addInstance发起请求的

public synchronized void setInstanceStatus(InstanceStatus status) {
        InstanceStatus next = instanceStatusMapper.map(status);
        if (next == null) {
            return;
        }

        InstanceStatus prev = instanceInfo.setStatus(next);
        if (prev != null) {
            for (StatusChangeListener listener : listeners.values()) {
                try {
                    listener.notify(new StatusChangeEvent(prev, next));
                } catch (Exception e) {
                    logger.warn("failed to notify listener: {}", listener.getId(), e);
                }
            }
        }
    }

DiscoveryClient类内部

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

指向

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

最后即是向addInstance发起请求的地方

    boolean register() throws Throwable {
        logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
        EurekaHttpResponse<Void> httpResponse;
        try {
            httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
        } catch (Exception e) {
            logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
            throw e;
        }
        if (logger.isInfoEnabled()) {
            logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());
        }
        return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
    }

发起请求即是向ApplicationResource的Instance方法发起。

eureka续约在客户端和服务端分别怎么操作实现可用性的

从上面注册中可推测出续约/心跳接口可能也是在DiscoveryClient中完成的。搜索HeatBeat之后发现注入 DiscoveryClient方法中有一个初始化定时任务的方法

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

其中就有心跳的定时任务。默认的心跳间隔时间renewalIntervalInSecs为30秒

    /**
     * The heartbeat task that renews the lease in the given intervals.
     */
    private class HeartbeatThread implements Runnable {

        public void run() {
            if (renew()) {
                lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
            }
        }
    }

renew方法中即是向服务端发起调用的过程，与上述注册基本相同

Eureka下线服务

Eureka下线是在EurekaClientAutoConfiguration中注入EurekaClient时定义的shutDown方法。我们可以看到

   @PreDestroy
    @Override
    public synchronized void shutdown() {
        if (isShutdown.compareAndSet(false, true)) {
            logger.info("Shutting down DiscoveryClient ...");

            if (statusChangeListener != null && applicationInfoManager != null) {
                applicationInfoManager.unregisterStatusChangeListener(statusChangeListener.getId());
            }

            cancelScheduledTasks();

            // If APPINFO was registered
            if (applicationInfoManager != null
                    && clientConfig.shouldRegisterWithEureka()
                    && clientConfig.shouldUnregisterOnShutdown()) {
                applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
                unregister();
            }

            if (eurekaTransport != null) {
                eurekaTransport.shutdown();
            }

            heartbeatStalenessMonitor.shutdown();
            registryStalenessMonitor.shutdown();

            logger.info("Completed shut down of DiscoveryClient");
        }
    }

同样的。执行了一个取消定时任务的状态。。另外利用上面说的applicationInfoManager.setInstanceStatus()方法进行了事件通知，另外unregister();进行了取消注册操作。eurekaTransport.shutdown();关闭传输。

Eureka的功能特性总体上来说就是这样。有些地方可能还是不够清楚。欢迎大家一起沟通探讨

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2021-07-09，如有侵权请联系 cloudcommunity@tencent.com 删除

linux

本文分享自微瞰技术微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

linux

登录后参与评论

0 条评论

热度