入口：ApplicationResource#addInstance

    1.  @POST
    2.     @Consumes({"application/json", "application/xml"})
    3.     public Response addInstance(InstanceInfo info,
    4.                                 @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
    5.         logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
    6.         // 校验客户端数据
    7.         if (isBlank(info.getId())) {
    8.             return Response.status(400).entity("Missing instanceId").build();
    9.         } else if (isBlank(info.getHostName())) {
    10.             return Response.status(400).entity("Missing hostname").build();
    11.         } else if (isBlank(info.getIPAddr())) {
    12.             return Response.status(400).entity("Missing ip address").build();
    13.         } else if (isBlank(info.getAppName())) {
    14.             return Response.status(400).entity("Missing appName").build();
    15.         } else if (!appName.equals(info.getAppName())) {
    16.             return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();
    17.         } else if (info.getDataCenterInfo() == null) {
    18.             return Response.status(400).entity("Missing dataCenterInfo").build();
    19.         } else if (info.getDataCenterInfo().getName() == null) {
    20.             return Response.status(400).entity("Missing dataCenterInfo Name").build();
    21.         }
    24.         // 处理客户端数据丢失情况
    25.         DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
    26.         if (dataCenterInfo instanceof UniqueIdentifier) {
    27.             String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
    28.             if (isBlank(dataCenterInfoId)) {
    29.                 boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
    30.                 if (experimental) {
    31.                     String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
    32.                     return Response.status(400).entity(entity).build();
    33.                 } else if (dataCenterInfo instanceof AmazonInfo) {
    34.                     AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
    35.                     String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);
    36.                     if (effectiveId == null) {
    37.                         amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());
    38.                     }
    39.                 } else {
    40.                     logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
    41.                 }
    42.             }
    43.         }
    45.         // 注册
    46.         // isReplication参数作用则是判断这是否是集群之间的复制。防止重复注册。
    47.         registry.register(info, "true".equals(isReplication));
    48.         return Response.status(204).build();  // 204 to be backwards compatible
    49.     }

    上面参数的InstanceInfo对象就是服务实例的具体信息，比如，IP，集群微服务名，实例id，端口，最后操作时间，心跳url等信息，代码如下

    1. public class InstanceInfo {
    3.     private static final String VERSION_UNKNOWN = "unknown";
    5.     public static final int DEFAULT_PORT = 7001;
    6.     public static final int DEFAULT_SECURE_PORT = 7002;
    7.     public static final int DEFAULT_COUNTRY_ID = 1; // US
    9.     // The (fixed) instanceId for this instanceInfo. This should be unique within the scope of the appName.
    10.     // 实例id
    11.     private volatile String instanceId;
    13.     // 集群微服务名
    14.     private volatile String appName;
    15.     @Auto
    16.     private volatile String appGroupName;
    18.     // IP地址
    19.     private volatile String ipAddr;
    21.     private static final String SID_DEFAULT = "na";
    22.     @Deprecated
    23.     private volatile String sid = SID_DEFAULT;
    25.     private volatile int port = DEFAULT_PORT;
    26.     private volatile int securePort = DEFAULT_SECURE_PORT;
    28.     @Auto
    29.     private volatile String homePageUrl;
    30.     @Auto
    31.     private volatile String statusPageUrl;
    32.     @Auto
    33.     private volatile String healthCheckUrl;
    34.     @Auto
    35.     private volatile String secureHealthCheckUrl;
    36.     @Auto
    37.     private volatile String vipAddress;
    38.     @Auto
    39.     private volatile String secureVipAddress;
    40.     @XStreamOmitField
    41.     private String statusPageRelativeUrl;
    42.     @XStreamOmitField
    43.     private String statusPageExplicitUrl;
    44.     @XStreamOmitField
    45.     private String healthCheckRelativeUrl;
    46.     @XStreamOmitField
    47.     private String healthCheckSecureExplicitUrl;
    48.     @XStreamOmitField
    49.     private String vipAddressUnresolved;
    50.     @XStreamOmitField
    51.     private String secureVipAddressUnresolved;
    52.     @XStreamOmitField
    53.     private String healthCheckExplicitUrl;
    54.     @Deprecated
    55.     private volatile int countryId = DEFAULT_COUNTRY_ID; // Defaults to US
    56.     private volatile boolean isSecurePortEnabled = false;
    57.     private volatile boolean isUnsecurePortEnabled = true;
    58.     private volatile DataCenterInfo dataCenterInfo;
    59.     private volatile String hostName;
    60.     private volatile InstanceStatus status = InstanceStatus.UP;
    61.     private volatile InstanceStatus overriddenStatus = InstanceStatus.UNKNOWN;
    62.     @XStreamOmitField
    63.     private volatile boolean isInstanceInfoDirty = false;
    64.     private volatile LeaseInfo leaseInfo;
    65.     @Auto
    66.     private volatile Boolean isCoordinatingDiscoveryServer = Boolean.FALSE;
    67.     @XStreamAlias("metadata")
    68.     private volatile Map<String, String> metadata;
    69.     @Auto
    70.     // 最后操作时间
    71.     private volatile Long lastUpdatedTimestamp;
    72.     @Auto
    73.     private volatile Long lastDirtyTimestamp;
    74.     @Auto
    75.     private volatile ActionType actionType;
    76.     @Auto
    77.     private volatile String asgName;
    78.     private String version = VERSION_UNKNOWN;

    关键代码类图
    image.png

    通过代码跟踪，发现关键代码在PeerAwareInstanceRegistryImpl类中的register方法，这个类主要负责集群信息同步

    1. @Override
    2. public void register(final InstanceInfo info, final boolean isReplication) {
    3.     // 默认的续约时间为90秒
    4.     int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
    5.     // 如果在配置文件中配置了续约时间则重置续约时间
    6.     if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
    7.         leaseDuration = info.getLeaseInfo().getDurationInSecs();
    8.     }
    9.     // 调用父类的注册方法
    10.     super.register(info, leaseDuration, isReplication);
    11.     // 同步集群中所有的节点
    12.     replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
    13. }

    父类AbstractInstanceRegistry的register方法

    1. public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
    2.         try {
    3.             read.lock();
    4.              // 通过传进来的集群微服务名拿到一个微服务组
    5.             Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
    6.             REGISTER.increment(isReplication);
    7.             if (gMap == null) {
    8.                 final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
    9.                 gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
    10.                 if (gMap == null) {
    11.                     gMap = gNewMap;
    12.                 }
    13.             }
    14.             // 通过实例id拿到租债器对象，existingLease 表示已经存在的微服务实例对象
    15.             // getHolder() 具体的微服务实例对象
    16.             Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
    17.             // Retain the last dirty timestamp without overwriting it, if there is already a lease
    18.             if (existingLease != null && (existingLease.getHolder() != null)) {
    19.                 Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
    20.                 Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
    21.                 logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
    23.                 // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
    24.                 // InstanceInfo instead of the server local copy.
    25.                 if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
    26.                     logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
    27.                             " than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
    28.                     logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
    29.                     registrant = existingLease.getHolder();
    30.                 }
    31.             } else {
    32.                 // The lease does not exist and hence it is a new registration
    33.                 synchronized (lock) {
    34.                     if (this.expectedNumberOfClientsSendingRenews > 0) {
    35.                         // Since the client wants to register it, increase the number of clients sending renews
    36.                         this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
    37.                         updateRenewsPerMinThreshold();
    38.                     }
    39.                 }
    40.                 logger.debug("No previous lease information found; it is new registration");
    41.             }
    42.             // 将传入的服务实例组装成一个lease租债器对象
    43.             // leaseDuration 续约时间
    44.             Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
    45.             if (existingLease != null) {
    46.                 lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
    47.             }
    48.             gMap.put(registrant.getId(), lease);
    49.             recentRegisteredQueue.add(new Pair<Long, String>(
    50.                     System.currentTimeMillis(),
    51.                     registrant.getAppName() + "(" + registrant.getId() + ")"));
    52.             // This is where the initial state transfer of overridden status happens
    53.             if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
    54.                 logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
    55.                                 + "overrides", registrant.getOverriddenStatus(), registrant.getId());
    56.                 if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
    57.                     logger.info("Not found overridden id {} and hence adding it", registrant.getId());
    58.                     overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
    59.                 }
    60.             }
    61.             InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
    62.             if (overriddenStatusFromMap != null) {
    63.                 logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
    64.                 registrant.setOverriddenStatus(overriddenStatusFromMap);
    65.             }
    67.             // Set the status based on the overridden status rules
    68.             InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
    69.             registrant.setStatusWithoutDirty(overriddenInstanceStatus);
    71.             // If the lease is registered with UP status, set lease service up timestamp
    72.             if (InstanceStatus.UP.equals(registrant.getStatus())) {
    73.                 lease.serviceUp();
    74.             }
    75.             registrant.setActionType(ActionType.ADDED);
    76.             recentlyChangedQueue.add(new RecentlyChangedItem(lease));
    77.             registrant.setLastUpdatedTimestamp();
    78.             invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
    79.             logger.info("Registered instance {}/{} with status {} (replication={})",
    80.                     registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
    81.         } finally {
    82.             read.unlock();
    83.         }
    84.     }

    服务注册主要逻辑是

    • 通过服务集群名拿到一个一个微服务组，如果微服务组不存在则创建一个；
    • 根据实例id判断判断微服务组中是否存在实例对象，主要是在并发情况下发生多次注册，如果存在则哪个时间戳最新就用哪个；
    • 根据传入的参数初始化一个租债器对象，将实例的id为key，value为租债器对象放入微服务组中。