Spring Cloud-1.1-Eureka核心源码学习

文章目录

Eureka Server启动过程
Eureka Server服务接口暴露策略
Eureka Server服务注册接⼝（接受客户端注册服务）
Eureka Server服务续约接⼝（接受客户端续约）
Eureka Client注册服务
Eureka Client下架服务
Eureka Client心跳续约

Eureka Server启动过程

⼊⼝：SpringCloud充分利⽤了SpringBoot的⾃动装配的特点

观察eureka-server的jar包，发现在META-INF下⾯有配置⽂件spring.factories

springboot应⽤启动时会加载EurekaServerAutoConfiguration⾃动配置类
EurekaServerAutoConfiguration类

⾸先观察类头分析

** 需要有⼀个marker bean**，才能装配Eureka Server，那么这个marker其实是由@EnableEurekaServer注解决定的

也就是说只有添加了@EnableEurekaServer注解，才会有后续的动作，这是成为⼀个EurekaServer的前提
关注EurekaServerAutoConfiguration

⽽在 com.netflix.eureka.cluster.PeerEurekaNodes#start⽅法中

回到主配置类中

回到主配置类中

关注EurekaServerInitializerConfiguration

重点关注，进⼊
org.springframework.cloud.netflix.eureka.server.EurekaServerBootstrap#contextInitialized

重点关注initEurekaServerContext()

研究⼀下上图中的syncUp⽅法
com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#syncUp

继续研究com.netflix.eureka.registry.AbstractInstanceRegistry#register（提供实例注册功能）

回到 org.springframework.cloud.netflix.eureka.server.EurekaServerBootstrap#initEurekaServerContext方法中
继续研究com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#openForTraffic

进⼊postInit()⽅法查看

Eureka Server服务接口暴露策略

在Eureka Server启动过程中主配置类注册了Jersey框架（是⼀个发布restful⻛格接⼝的框架，类似于我们的springmvc）

注⼊的Jersey细节

扫描classpath下的那些packages呢？已经定义好了

对外提供的接⼝服务，在Jersey中叫做资源

这些就是使⽤Jersey发布的供Eureka Client调⽤的Restful⻛格服务接⼝（完成服务注册、⼼跳续约等接⼝）

Eureka Server服务注册接⼝（接受客户端注册服务）

ApplicationResource类的addInstance()⽅法中代码：registry.register(info,"true".equals(isReplication));

com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#register - 注册服务信息并同步到其它Eureka节点

AbstractInstanceRegistry#register()：注册，实例信息存储到注册表是⼀个ConcurrentHashMap

    /**
     * Registers a new instance with a given duration.
     *
     * @see com.netflix.eureka.lease.LeaseManager#register(java.lang.Object, int, boolean)
     */
    public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
   
        read.lock();//读锁
        try {
   
        	// registry是保存所有应⽤实例信息的Map：ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>
			// 从registry中获取当前appName的所有实例信息
            Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
            REGISTER.increment(isReplication);//注册统计+1
            // 如果当前appName实例信息为空，新建Map
            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;
                }
            }
            // 获取实例的Lease租约信息
            Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
            // Retain the last dirty timestamp without overwriting it, if there is already a lease
            // 如果已经有租约，则保留最后⼀个脏时间戳⽽不覆盖它
			// （⽐较当前请求实例租约 和 已有租约 的LastDirtyTimestamp，选择靠后的）
            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
                // 如果之前不存在实例的租约，说明是新实例注册
                // expectedNumberOfRenewsPerMin期待的每分钟续约数+2（因为30s⼀个）
				// 并更新numberOfRenewsPerMinThreshold每分钟续约阀值（85%）
                synchronized (lock) {
   
                    if (this.expectedNumberOfClientsSendingRenews > 0) {
   
                        // Since the client wants to register it, increase the number of clients sending renews
                        this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
                        updateRenewsPerMinThreshold();
                    }
                }
                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());
            }
            gMap.put(registrant.getId(), lease);
            //当前实例信息放到维护注册信息的Map
			// 同步维护最近注册队列
            recentRegisteredQueue.add(new Pair<Long, String>(
                    System.currentTimeMillis(),
                    registrant.getAppName() + "(" + registrant.getId() + ")"));
            // This is where the initial state transfer of overridden status happens
            // 如果当前实例已经维护了OverriddenStatus，将其也放到此EurekaServer的overriddenInstanceStatusMap中
            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());
                }
            }
            // 根据overridden status规则，设置状态
            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
            // 如果租约以UP状态注册，设置租赁服务时间戳
            if (InstanceStatus.UP.equals(registrant.getStatus())) {
   
                lease.serviceUp();
            }
            registrant.setActionType(ActionType.ADDED);//ActionType为ADD
            recentlyChangedQueue.add(new RecentlyChangedItem(lease));//维护recentlyChangedQueue
            registrant.setLastUpdatedTimestamp();//更新最后更新时间
            // 使当前应⽤的ResponseCache失效
            invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
            logger.info("Registered instance {}/{} with status {} (replication={})",
                    registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
        } finally {
   
            read.unlock();//释放锁
        }
    }

PeerAwareInstanceRegistryImpl#replicateToPeers() ：复制到Eureka对等节点

    /**
     * Replicates all eureka actions to peer eureka nodes except for replication
     * traffic to this node.
     *
     */
    private void replicateToPeers(Action action, String appName, String id,
                                  InstanceInfo info /* optional */,
                                  InstanceStatus newStatus /* optional */, boolean isReplication) {
   
        Stopwatch tracer = action.getTimer().start();
        try {
   
       		 // 如果是复制操作（针对当前节点，false）
            if (isReplication) {
   
                numberOfReplicationsLastMin.increment();
            }
            // If it is a replication already, do not replicate again as this will create a poison replication
            // 如果它已经是复制，请不要再次复制，直接return
            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;
                }
                // 复制Instance实例操作到某个node节点
                replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
            }
        } finally {
   
            tracer.stop();
        }
    }

PeerAwareInstanceRegistryImpl#replicateInstanceActionsToPeers

Eureka Server服务续约接⼝（接受客户端续约）

InstanceResource的renewLease⽅法中完成客户端的⼼跳（续约）处理，关键代码：registry.renew(app.getName(), id, isFromReplicaNode);

com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#renew

replicateInstanceActionsToPeers() 复制Instance实例操作到其它节点, 就是调用到上面的

enew()⽅法中--->leaseToRenew.renew()--->对最后更新时间戳进⾏更新

Eureka Client注册服务

启动过程：Eureka客户端在启动时也会装载很多配置类，我们通过spring-cloudnetflix-eureka-client-2.7.7.RELEASE.jar下的spring.factories⽂件可以看到加载的配置类

引⼊jar就会被⾃动装配，分析EurekaClientAutoConfiguration类头

回到主配置类EurekaClientAutoConfiguration
思考：EurekaClient启动过程要做什么事情？？？？？？

读取配置⽂件
启动时从EurekaServer获取服务实例信息

观察⽗类DiscoveryClient()，最终来到com.netflix.discovery.DiscoveryClient#DiscoveryClient(com.netflix.appinfo.ApplicationInfoManager, com.netflix.discovery.EurekaClientConfig, com.netflix.discovery.AbstractDiscoveryClientOptionalArgs, javax.inject.Provider<com.netflix.discovery.BackupRegistry>, com.netflix.discovery.shared.resolver.EndpointRandomizer)

再到com.netflix.discovery.DiscoveryClient#fetchRegistry

注册⾃⼰到EurekaServer（addInstance）
回到DiscoveryClient构造器

DiscoveryClient#register

底层使⽤Jersey客户端进⾏远程请求。
开启⼀些定时任务（⼼跳续约，刷新本地服务缓存列表）

⼼跳续约定时任务

来到com.netflix.eureka.transport.JerseyReplicationClient#sendHeartBeat