springCloud-Eureka源码分析

Eureka作为springCloud的注册中心，提供了服务注册、服务续约、服务同步等功能，本片文章结合源码来看下Eureka核心功能

Eureka架构图

Eureka核心功能

• 服务注册(register)：Eureka Client会通过发送REST请求的方式向Eureka Server注册自己的服务，提供自身的元数
  据，比如ip地址、端口、运行状况指标的url、主页地址等信息。Eureka Server接收到注册请求后，就会把这些元数
  据信息存储在一个双层的Map中。
• 服务续约(renew)：在服务注册后，Eureka Client会维护一个心跳来持续通知Eureka Server，说明服务一直处于可
  用状态，防止被剔除。Eureka Client在默认的情况下会每隔30秒(eureka.instance.leaseRenewallIntervalInSeconds)发送一次心跳来进行服务续约。
• 服务同步(replicate)：Eureka Server之间会互相进行注册，构建Eureka Server集群，不同Eureka Server之间会进
  行服务同步，用来保证服务信息的一致性。
• 获取服务(get registry)：服务消费者（Eureka Client）在启动的时候，会发送一个REST请求给Eureka Server，获
  取上面注册的服务清单，并且缓存在Eureka Client本地，默认缓存30秒
  (eureka.client.registryFetchIntervalSeconds)。同时，为了性能考虑，Eureka Server也会维护一份只读的服务清
  单缓存，该缓存每隔30秒更新一次。
• 服务调用：服务消费者在获取到服务清单后，就可以根据清单中的服务列表信息，查找到其他服务的地址，从而进行
  远程调用。Eureka有Region和Zone的概念，一个Region可以包含多个Zone，在进行服务调用时，优先访问处于同
  一个Zone中的服务提供者。
• 服务下线(cancel)：当Eureka Client需要关闭或重启时，就不希望在这个时间段内再有请求进来，所以，就需要提前
  先发送REST请求给Eureka Server，告诉Eureka Server自己要下线了，Eureka Server在收到请求后，就会把该服务
  状态置为下线（DOWN），并把该下线事件传播出去。
• 服务剔除(evict)：有时候，服务实例可能会因为网络故障等原因导致不能提供服务，而此时该实例也没有发送请求给
  Eureka Server来进行服务下线，所以，还需要有服务剔除的机制。Eureka Server在启动的时候会创建一个定时任
  务，每隔一段时间（默认60秒），从当前服务清单中把超时没有续约（默认90秒，
  eureka.instance.leaseExpirationDurationInSeconds）的服务剔除。
• 自我保护：既然Eureka Server会定时剔除超时没有续约的服务，那就有可能出现一种场景，网络一段时间内发生了
  异常，所有的服务都没能够进行续约，Eureka Server就把所有的服务都剔除了，这样显然不太合理。所以，就有了
  自我保护机制，当短时间内，统计续约失败的比例，如果达到一定阈值，则会触发自我保护的机制，在该机制下，
  Eureka Server不会剔除任何的微服务，等到正常后，再退出自我保护机制。自我保护开关(eureka.server.enable-
  self-preservation: false)

##Eureka Server端源码分析

EurekaServerAutoConfiguration

筛选了部分核心代码进行说明

@Configuration(
    proxyBeanMethods = false
)
@Import({EurekaServerInitializerConfiguration.class})
@ConditionalOnBean({Marker.class})
@EnableConfigurationProperties({EurekaDashboardProperties.class, InstanceRegistryProperties.class})
@PropertySource({"classpath:/eureka/server.properties"})
public class EurekaServerAutoConfiguration implements WebMvcConfigurer {
    private static final String[] EUREKA_PACKAGES = new String[]{"com.netflix.discovery", "com.netflix.eureka"};
    @Autowired
    private ApplicationInfoManager applicationInfoManager;
    @Autowired
    private EurekaServerConfig eurekaServerConfig;
    @Autowired
    private EurekaClientConfig eurekaClientConfig;
    @Autowired
    private EurekaClient eurekaClient;
    @Autowired
    private InstanceRegistryProperties instanceRegistryProperties;
    public static final CloudJacksonJson JACKSON_JSON = new CloudJacksonJson();

    public EurekaServerAutoConfiguration() {
    }

    @Bean
    public HasFeatures eurekaServerFeature() {
        return HasFeatures.namedFeature("Eureka Server", EurekaServerAutoConfiguration.class);
    }

  // 加载EurekaController, spring‐cloud 提供了一些额外的接口，用来获取eurekaServer的信息
    @Bean
    @ConditionalOnProperty(
        prefix = "eureka.dashboard",
        name = {"enabled"},
        matchIfMissing = true
    )
    public EurekaController eurekaController() {
        return new EurekaController(this.applicationInfoManager);
    }

    @Bean
    public ServerCodecs serverCodecs() {
        return new EurekaServerAutoConfiguration.CloudServerCodecs(this.eurekaServerConfig);
    }

    private static CodecWrapper getFullJson(EurekaServerConfig serverConfig) {
        CodecWrapper codec = CodecWrappers.getCodec(serverConfig.getJsonCodecName());
        return codec == null ? CodecWrappers.getCodec(JACKSON_JSON.codecName()) : codec;
    }

    private static CodecWrapper getFullXml(EurekaServerConfig serverConfig) {
        CodecWrapper codec = CodecWrappers.getCodec(serverConfig.getXmlCodecName());
        return codec == null ? CodecWrappers.getCodec(XStreamXml.class) : codec;
    }

    @Bean
    @ConditionalOnMissingBean
    public ReplicationClientAdditionalFilters replicationClientAdditionalFilters() {
        return new ReplicationClientAdditionalFilters(Collections.emptySet());
    }

  //初始化集群注册表
    @Bean
    public PeerAwareInstanceRegistry peerAwareInstanceRegistry(ServerCodecs serverCodecs) {
        this.eurekaClient.getApplications();
        return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig, serverCodecs, this.eurekaClient, this.instanceRegistryProperties.getExpectedNumberOfClientsSendingRenews(), this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
    }

   // 配置服务节点信息，这里的作用主要是为了配置Eureka的peer节点，也就是说当有收到有节点注册上来 
    //的时候，需要通知给那些服务节点， （互为一个集群）
    @Bean
    @ConditionalOnMissingBean
    public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry, ServerCodecs serverCodecs, ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
        return new EurekaServerAutoConfiguration.RefreshablePeerEurekaNodes(registry, this.eurekaServerConfig, this.eurekaClientConfig, serverCodecs, this.applicationInfoManager, replicationClientAdditionalFilters);
    }

  //EurekaServer上下文
    @Bean
    @ConditionalOnMissingBean
    public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs, PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
        return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs, registry, peerEurekaNodes, this.applicationInfoManager);
    }

  // 这个类的作用是spring‐cloud和原生eureka的胶水代码，通过这个类来启动EurekaSever 
  // 后面这个类会在EurekaServerInitializerConfiguration被调用，进行eureka启动
    @Bean
    public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry, EurekaServerContext serverContext) {
        return new EurekaServerBootstrap(this.applicationInfoManager, this.eurekaClientConfig, this.eurekaServerConfig, registry, serverContext);
    }

    // 配置拦截器，ServletContainer里面实现了jersey框架，通过他来实现eurekaServer对外的restFull接口
    @Bean
    public FilterRegistrationBean<?> jerseyFilterRegistration(Application eurekaJerseyApp) {
        FilterRegistrationBean<Filter> bean = new FilterRegistrationBean();
        bean.setFilter(new ServletContainer(eurekaJerseyApp));
        bean.setOrder(2147483647);
        bean.setUrlPatterns(Collections.singletonList("/eureka/*"));
        return bean;
    }
}

EurekaServerInitializerConfiguration

EurekaServerAutoConfiguration会导入EurekaServerInitializerConfiguration

@Configuration(
    proxyBeanMethods = false
)
public class EurekaServerInitializerConfiguration implements ServletContextAware, SmartLifecycle, Ordered {
    private static final Log log = LogFactory.getLog(EurekaServerInitializerConfiguration.class);
    @Autowired
    private EurekaServerConfig eurekaServerConfig;
    private ServletContext servletContext;
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private EurekaServerBootstrap eurekaServerBootstrap;
    private boolean running;
    private int order = 1;

    public EurekaServerInitializerConfiguration() {
    }

    public void setServletContext(ServletContext servletContext) {
        this.servletContext = servletContext;
    }

     //启动一个线程
    public void start() {
        (new Thread(() -> {
            try {
                //初始化EurekaServer，同时注册Eureka Server
                this.eurekaServerBootstrap.contextInitialized(this.servletContext);
                log.info("Started Eureka Server");
                //发布EurekaServer注册事件
                this.publish(new EurekaRegistryAvailableEvent(this.getEurekaServerConfig()));
                // 设置启动的状态为true
                this.running = true;
                // 发送Eureka Start 事件 ， 其他还有各种事件，我们可以监听这种时间，然后做一些特定的业务需求
                this.publish(new EurekaServerStartedEvent(this.getEurekaServerConfig()));
            } catch (Exception var2) {
                log.error("Could not initialize Eureka servlet context", var2);
            }

        })).start();
    }


    private EurekaServerConfig getEurekaServerConfig() {
        return this.eurekaServerConfig;
    }

    private void publish(ApplicationEvent event) {
        this.applicationContext.publishEvent(event);
    }

    public void stop() {
        this.running = false;
        this.eurekaServerBootstrap.contextDestroyed(this.servletContext);
    }

    public boolean isRunning() {
        return this.running;
    }

    public int getPhase() {
        return 0;
    }

    public boolean isAutoStartup() {
        return true;
    }

    public void stop(Runnable callback) {
        callback.run();
    }

    public int getOrder() {
        return this.order;
    }
}

EurekaServerBootstrap

EurekaServerBootstrap的contextInitialized初始化方法

public class EurekaServerBootstrap {
  //初始化EurekaServer的运行环境和上下文
    public void contextInitialized(ServletContext context) {
        try {
            this.initEurekaEnvironment();
            this.initEurekaServerContext();
            context.setAttribute(EurekaServerContext.class.getName(), this.serverContext);
        } catch (Throwable var3) {
            log.error("Cannot bootstrap eureka server :", var3);
            throw new RuntimeException("Cannot bootstrap eureka server :", var3);
        }
    }

    
   //初始化EurekaServer的上下文
    protected void initEurekaServerContext() throws Exception {
        JsonXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(), 10000);
        XmlXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(), 10000);
        if (this.isAws(this.applicationInfoManager.getInfo())) {
            this.awsBinder = new AwsBinderDelegate(this.eurekaServerConfig, this.eurekaClientConfig, this.registry, this.applicationInfoManager);
            this.awsBinder.start();
        }
       //初始化eureka server上下文
        EurekaServerContextHolder.initialize(this.serverContext);
        log.info("Initialized server context");
       // 从相邻的eureka节点复制注册表
        int registryCount = this.registry.syncUp();
      // 默认每30秒发送心跳，1分钟就是2次 38 
      // 修改eureka状态为up 39
      // 同时，这里面会开启一个定时任务，用于清理60秒没有心跳的客户端。自动下线       
        this.registry.openForTraffic(this.applicationInfoManager, registryCount);
        EurekaMonitors.registerAllStats();
    }

    protected void destroyEurekaServerContext() throws Exception {
        EurekaMonitors.shutdown();
        if (this.awsBinder != null) {
            this.awsBinder.shutdown();
        }

        if (this.serverContext != null) {
            this.serverContext.shutdown();
        }

    }

    protected void destroyEurekaEnvironment() throws Exception {
    }

    protected boolean isAws(InstanceInfo selfInstanceInfo) {
        boolean result = Name.Amazon == selfInstanceInfo.getDataCenterInfo().getName();
        log.info("isAws returned " + result);
        return result;
    }
}


public int syncUp() {
        int count = 0;

        for(int i = 0; i < this.serverConfig.getRegistrySyncRetries() && count == 0; ++i) {
            if (i > 0) {
                try {
                    Thread.sleep(this.serverConfig.getRegistrySyncRetryWaitMs());
                } catch (InterruptedException var10) {
                    logger.warn("Interrupted during registry transfer..");
                    break;
                }
            }

            Applications apps = this.eurekaClient.getApplications();
            Iterator var4 = apps.getRegisteredApplications().iterator();

            while(var4.hasNext()) {
                Application app = (Application)var4.next();
                Iterator var6 = app.getInstances().iterator();

                while(var6.hasNext()) {
                    InstanceInfo instance = (InstanceInfo)var6.next();

                    try {
                        if (this.isRegisterable(instance)) {
                            //将其他节点的实例注册到本节点
                            this.register(instance, instance.getLeaseInfo().getDurationInSecs(), true);
                            ++count;
                        }
                    } catch (Throwable var9) {
                        logger.error("During DS init copy", var9);
                    }
                }
            }
        }

        return count;
    }



 public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
         // 计算每分钟最大续约数
        this.expectedNumberOfClientsSendingRenews = count;
        this.updateRenewsPerMinThreshold();
        logger.info("Got {} instances from neighboring DS node", count);
        logger.info("Renew threshold is: {}", this.numberOfRenewsPerMinThreshold);
        this.startupTime = System.currentTimeMillis();
        if (count > 0) {
            this.peerInstancesTransferEmptyOnStartup = false;
        }

        Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
        boolean isAws = Name.Amazon == selfName;
        if (isAws && this.serverConfig.shouldPrimeAwsReplicaConnections()) {
            logger.info("Priming AWS connections for all replicas..");
            this.primeAwsReplicas(applicationInfoManager);
        }

        logger.info("Changing status to UP");
        //设置实例状态为up
        applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
        // 开启定时任务，默认60秒执行一次，用于清理60秒之内没有续约的实例
        super.postInit();
    }

从上面的EurekaServerAutoConfiguration类，我们可以看到有个初始化EurekaServerContext的方法

@Bean
@ConditionalOnMissingBean
public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs, PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
    return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs, registry, peerEurekaNodes, this.applicationInfoManager);
}

DefaultEurekaServerContext 这个类里面的的initialize()方法是被@PostConstruct 这个注解修饰的,

在应用加载的时候，会执行这个方法

@PostConstruct
 public void initialize() {
     logger.info("Initializing ...");
     // 启动一个线程，读取其他集群节点的信息，后面后续复制
     this.peerEurekaNodes.start();

     try {
         this.registry.init(this.peerEurekaNodes);
     } catch (Exception var2) {
         throw new RuntimeException(var2);
     }

     logger.info("Initialized");
 }

peerEurekaNodes.start()主要是启动一个只拥有一个线程的线程池，第一次进去会更新一下集群其他节点信息 然后启动了一个定时线程，每60秒更新一次，也就是说后续可以根据配置动态的修改节点配置。（原生的spring cloud config支持）

public void start() {
        this.taskExecutor = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                thread.setDaemon(true);
                return thread;
            }
        });

        try {
            // 首次进来，更新集群节点信息
            this.updatePeerEurekaNodes(this.resolvePeerUrls());
            //搞个线程
            Runnable peersUpdateTask = new Runnable() {
                public void run() {
                    try {
                        PeerEurekaNodes.this.updatePeerEurekaNodes(PeerEurekaNodes.this.resolvePeerUrls());
                    } catch (Throwable var2) {
                        PeerEurekaNodes.logger.error("Cannot update the replica Nodes", var2);
                    }

                }
            };
            this.taskExecutor.scheduleWithFixedDelay(peersUpdateTask, (long)this.serverConfig.getPeerEurekaNodesUpdateIntervalMs(), (long)this.serverConfig.getPeerEurekaNodesUpdateIntervalMs(), TimeUnit.MILLISECONDS);
        } catch (Exception var3) {
            throw new IllegalStateException(var3);
        }

        Iterator var4 = this.peerEurekaNodes.iterator();

        while(var4.hasNext()) {
            PeerEurekaNode node = (PeerEurekaNode)var4.next();
            logger.info("Replica node URL:  {}", node.getServiceUrl());
        }

    }
  
  
  
// 根据URL 构建PeerEurekaNode信息
protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
        HttpReplicationClient replicationClient = JerseyReplicationClient.createReplicationClient(this.serverConfig, this.serverCodecs, peerEurekaNodeUrl);
        String targetHost = hostFromUrl(peerEurekaNodeUrl);
        if (targetHost == null) {
            targetHost = "host";
        }

        return new PeerEurekaNode(this.registry, targetHost, peerEurekaNodeUrl, replicationClient, this.serverConfig);
    }

Eureka Server留存图