Eureka服务端源码解析

一、自动装配

1、找到spring-cloud-starter-netflix-eureka-server.jar的spring.factories,查看spring.factories如下:

Eureka服务端源码解析

2、进入EurekaServer的自动装配类EurekaServerAutoConfiguration:

Eureka服务端源码解析

  3、@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)也就是说当容器中有EurekaServerMarkerConfig uration.Marker.class时,该配置类才起作用。接下来看下启动类EurekaApplication,该启动类上不光有@SpringBootApplication自动装配,还有@EnableEurekaServer,开启EurekaServer

二、EurekaServer 

1、@EnableEurekaServer注解开启EurekaServer 

  1.1 点进去@EnableEurekaServer点进去,发现其@Import(EurekaServerMarkerConfiguration.class),导入了EurekaServer MarkerConfiguration配置类

Eureka服务端源码解析

  1.2 EurekaServerMarkerConfiguration配置类,该配置类导入了EurekaServerMarkerConfiguration.Marker.class。如下:

Eureka服务端源码解析

   2、EurekaServerAutoConfiguration配置类

  当容器中有EurekaServerMarkerConfiguration.Marker.class,就可以激活该配置类,接下来详细看下该配置类为我们做了什么。

  2.1 @Import(EurekaServerInitializerConfiguration.class)

  

Eureka服务端源码解析

   EurekaServerInitializerConfiguration配置类实现了SmartLifecycle,我们知道实现了SmartLifecycle接口的,会在Ioc容器中所有Bean初始化完成后,根据isAutoStartup()方法返回true来执行该配置类的start()

  ① 进入EurekaServerInitializerConfiguration.start()方法:

public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    //EurekaServerAutoConfiguration->@Bean EurekaServerBootstrap
                    eurekaServerBootstrap.contextInitialized(
                            EurekaServerInitializerConfiguration.this.servletContext);
                    log.info("Started Eureka Server");
                    //发布EurekaRegistryAvailableEvent事件
                    publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
                    //设置运行状态为true
                    EurekaServerInitializerConfiguration.this.running = true;
                    //发布EurekaServerStartedEvent事件
                    publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
                }
                catch (Exception ex) {
                    // Help!
                    log.error("Could not initialize Eureka servlet context", ex);
                }
            }
        }).start();
    }

  ② 进入EurekaServerBootstrap.contextInitialized(ServletContext context)方法:

public void contextInitialized(ServletContext context) {
        try {
            //初始化EurekaServer的运行环境
            initEurekaEnvironment();
            //初始化EurekaServer的上下文
            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);
        }
    }

  ③ 进入EurekaServerBootstrap.initEurekaServerContext()方法:

protected void initEurekaServerContext() throws Exception {
        //省略非核心代码
        //从集群其他节点复制注册
        int registryCount = this.registry.syncUp();
        /**
         * 1、修改状态为UP
         * 2、调用父类的postInit 开启一个剔除定时任务,每隔60执行一次,从当前服务清单中把超时(默认90秒)没有续约剔
         */
        this.registry.openForTraffic(this.applicationInfoManager, registryCount);
        // Register all monitoring statistics.
        EurekaMonitors.registerAllStats();
    }

  ④ PeerAwareInstanceRegistryImpl.syncUp()方法:

  该方法中的eurekaClient.getApplications()是通过http调用,获取集群中的其他节点的所有服务实例。然后遍历获取到的apps,根据isRegisterable(instance)判断是否可注册,如果可以注册就调用register(instance, instance.getLeaseInfo().getDurationInSecs(), true)进行注册,注册实质就是往AbstractInstanceRegistry的属性private final ConcurrentHashMap<String, Map<String, Lease <InstanceInfo>>> registry = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();中加入服务实例信息。

  ⑤ PeerAwareInstanceRegistryImpl.openForTraffic()方法:

  该方法核心第一步:applicationInfoManager.setInstanceStatus(InstanceStatus.UP)修改状态为UP,第二步:调用super.postInit();开启一个剔除定时任务,每隔60执行一次,从当前服务清单中把超时(默认90秒)没有续约剔除。

  ⑥ 进入AbstractInstanceRegistry.postInit()方法:

protected void postInit() {
        renewsLastMin.start();
        if (evictionTaskRef.get() != null) {
            evictionTaskRef.get().cancel();
        }
        //设置剔除任务EvictionTask
        evictionTaskRef.set(new AbstractInstanceRegistry.EvictionTask());
        //每隔60s执行一次EvictionTask的run方法
        evictionTimer.schedule(evictionTaskRef.get(),
                serverConfig.getEvictionIntervalTimerInMs(),
                serverConfig.getEvictionIntervalTimerInMs());
    }

  ⑦ EvictionTask的run方法:

  执行AbstractInstanceRegistry.evict(),剔除逻辑:主要的功能是将注册表registry,其实就是一个ConcurrentHashMap的所有注册实例遍历下,看哪些是过期的,过期了就加入到expiredLeases中,然后遍历expiredLeases,执行internalCancel方法把实例状态修改成DELETED状态,这样客户端就拿不到。

  2.2 导入的核心Bean

  ① EurekaServerConfig:初始化eurekaServer配置;

  ② EurekaController:初始化dashboard的相关接口,用户获取eurekaServer的相关信息;

  ③ PeerAwareInstanceRegistry:初始化集群注册表;

  ④ PeerEurekaNodes:初始化集群节点;

  ⑤ EurekaServerContext:基于eurekaServer配置,注册表,集群节点,以及服务实例初始化eurekaServer上下文;

  ⑥ EurekaServerBootstrap:初始化eureka启动类;

  ⑦ FilterRegistrationBean:往Filter注册表里面注册一个Jsrsey过滤器;

  其中EurekaServerContext的默认实现DefaultEurekaServerContext在初始化的时候会调用initialize()方法,流程图如下:

Eureka服务端源码解析

   3、Eureka的Jersey服务

  3.1 服务注册接口

  ApplicationResource.addInstance()方法,核心逻辑就是调用PeerAwareInstanceRegistryImpl.register(final InstanceInfo info, final boolean isReplication)方法如下:

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();
  }
  //往注册表中注册实例信息,然后执行invalidateCache(),把读写缓存readWriteCacheMap失效掉
  super.register(info, leaseDuration, isReplication);
  //复制到集群中的其他节 发起http调用,调用集群中的其他节点的注册服务实例接口
  replicateToPeers(PeerAwareInstanceRegistryImpl.Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
}

  3.2 获取全量实例信息接口

  ApplicationsResource.getContainers()方法如下:

public Response getContainers(@PathParam("version") String version,
     @HeaderParam(HEADER_ACCEPT) String acceptHeader,
     @HeaderParam(HEADER_ACCEPT_ENCODING) String acceptEncoding,
     @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept,
     @Context UriInfo uriInfo,
     @Nullable @QueryParam("regions") String regionsStr) {
     //省略......
     //1、构建缓存key
     Key cacheKey = new Key(Key.EntityType.Application,
                ResponseCacheImpl.ALL_APPS,
                keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
        );
        Response response;
        if (acceptEncoding != null && acceptEncoding.contains(HEADER_GZIP_VALUE)) {
            //省略......
        } else {
            //2、根据缓存key从缓存中获取,首先从只读缓存中取为null,再去读写缓存中取,然后设置到只读缓存中
            response = Response.ok(responseCache.get(cacheKey))
                    .build();
        }
        return response;
    }

  3.3 获取增量实例信息接口

  ApplicationsResource.getContainerDifferential()方法,逻辑同获取全量实例信息接口一样,不同在于构建缓存key的时候传入的ALL_APPS_DELTA,而获取全量实例信息接口传入的是ALL_APPS。

  3.4 心跳接口

  InstanceResource.renewLease()方法,核心逻辑就是调用PeerAwareInstanceRegistryImpl.renew(final String appName, final String id, final boolean isReplication)方法进行续约,其实就是设置实例的lastUpdateTimestamp为当前时间+duration

public void renew() {
   //设置lastUpdateTimestamp为当前时间+duration
   lastUpdateTimestamp = System.currentTimeMillis() + duration;
}

  3.5 服务下线接口

  InstanceResource.cancelLease()方法,核心就是调用PeerAwareInstanceRegistryImpl.cancel(final String appName, final String id,final boolean isReplication)方法进行服务下线,其实就是把实例的状态设置成DELETE,然后执行invalidateCache(),把读写缓存readWriteCacheMap失效掉

三、Eureka服务端流程图

Eureka服务端源码解析

上一篇:Spring cloud入门-4:服务注册中心-Eureka集群环境构建


下一篇:「环境搭建」SpringCloud环境搭建以及常用组件