九.SpringCloud源码剖析-Eureka Server服务注册流程

妖狐艹你老母 2022-12-11 06:29 319阅读 0赞

系列文章目录

一.SpringCloud源码剖析-Eureka核心API

二.SpringCloud源码剖析-Eureka Client 初始化过程

三.SpringCloud源码剖析-Eureka服务注册

四.SpringCloud源码剖析-Eureka服务发现

五.SpringCloud源码剖析-Eureka Client服务续约

六.SpringCloud源码剖析-Eureka Client取消注册

七.SpringCloud源码剖析-Eureka Server的自动配置

八.SpringCloud源码剖析-Eureka Server初始化流程

九.SpringCloud源码剖析-Eureka Server服务注册流程

十.SpringCloud源码剖析-Eureka Server服务续约

十一.SpringCloud源码剖析-Eureka Server服务注册表拉取

十二.SpringCloud源码剖析-Eureka Server服务剔除

十三.SpringCloud源码剖析-Eureka Server服务下线

前言

本片文章的目的是分析Eureka Server的注册流程，您可以结合《Eureka Client服务注册》更容易理解

Eureka Server服务注册流程

在《Eureka Server初始化流程》 文章中我们知道，在EurekaServerAutoConfiguration中注册了JerseyFilter用来处理所有的/eureka开头的请求,当Eureka Client客户端发起注册请求，请求被该Filter接待

  1. /** * Register the Jersey filter * 注册Jersey filter */
  2.     @Bean
  3.     public FilterRegistrationBean jerseyFilterRegistration(
  4.             javax.ws.rs.core.Application eurekaJerseyApp) { 
  5.         FilterRegistrationBean bean = new FilterRegistrationBean();
  6.         //ServletContainer 是核心处理类
  7.         bean.setFilter(new ServletContainer(eurekaJerseyApp));
  8.         bean.setOrder(Ordered.LOWEST_PRECEDENCE);
  9.         //处理的请求/eureka/**
  10.         bean.setUrlPatterns(
  11.                 Collections.singletonList(EurekaConstants.DEFAULT_PREFIX + "/*"));
  13.         return bean;
  14.     }

具体的实现是在ServletContainer中完成

  1. //处理请求
  2. private void doFilter(HttpServletRequest request, HttpServletResponse response, FilterChain chain, String requestURI, String servletPath, String queryString) throws IOException, ServletException { 
  3.         Pattern p = this.getStaticContentPattern();
  4.         //处理 ContextPath 上下文路径
  5.         if (p != null && p.matcher(servletPath).matches()) { 
  6.             chain.doFilter(request, response);
  7.         } else { 
  8.             if (this.filterContextPath != null) { 
  9.                 if (!servletPath.startsWith(this.filterContextPath)) { 
  10.                     throw new ContainerException("The servlet path, \"" + servletPath + "\", does not start with the filter context path, \"" + this.filterContextPath + "\"");
  11.                 }
  13.                 if (servletPath.length() == this.filterContextPath.length()) { 
  14.                     if (this.webComponent.getResourceConfig().getFeature("com.sun.jersey.config.feature.Redirect")) { 
  15.                         URI l = UriBuilder.fromUri(request.getRequestURL().toString()).path("/").replaceQuery(queryString).build(new Object[0]);
  16.                         response.setStatus(307);
  17.                         response.setHeader("Location", l.toASCIIString());
  18.                         return;
  19.                     }
  21.                     requestURI = requestURI + "/";
  22.                 }
  23.             }
  25.             UriBuilder absoluteUriBuilder = UriBuilder.fromUri(request.getRequestURL().toString());
  26.             URI baseUri = this.filterContextPath == null ? absoluteUriBuilder.replacePath(request.getContextPath()).path("/").build(new Object[0]) : absoluteUriBuilder.replacePath(request.getContextPath()).path(this.filterContextPath).path("/").build(new Object[0]);
  27.             URI requestUri = absoluteUriBuilder.replacePath(requestURI).replaceQuery(queryString).build(new Object[0]);
  28.             //这里调用service方法
  29.             int status = this.service(baseUri, requestUri, request, response);
  30.             if (this.forwardOn404 && status == 404 && !response.isCommitted()) { 
  31.                 response.setStatus(200);
  32.                 chain.doFilter(request, response);
  33.             }
  35.         }
  36.     }
  39.     // 处理请求，调用WebComponent的service方法
  40.     public int service(URI baseUri, URI requestUri, HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException { 
  41.         return this.webComponent.service(baseUri, requestUri, request, response);
  42.     }

最终会调用ApplicationResource实现服务注册 ， Eureka Server对于Eureka client注册服务实例，获取服务实例的的REST请求的都交给ApplicationResource处理，其中用来服务注册的方法是addInstance,我们来看一下他的源码

  1. /** 服务实例注册,InstanceInfo是服务注册信息，isReplicationd为true代表是从其他Eureka Server节点复制实例，如果是isReplication为false，代表是Eureka Client 注册的 * Registers information about a particular instance for an * {@link com.netflix.discovery.shared.Application}. * * @param info * {@link InstanceInfo} information of the instance. * @param isReplication * a header parameter containing information whether this is * replicated from other nodes. */
  2.     @POST
  3.     @Consumes({ "application/json", "application/xml"})
  4.     public Response addInstance(InstanceInfo info,
  5.                                 @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) { 
  6.         logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
  7.         //验证instanceinfo包含所有必需的必填字段
  8.         // validate that the instanceinfo contains all the necessary required fields
  9.         if (isBlank(info.getId())) { 
  10.             return Response.status(400).entity("Missing instanceId").build();
  11.         } else if (isBlank(info.getHostName())) { 
  12.             return Response.status(400).entity("Missing hostname").build();
  13.         } else if (isBlank(info.getIPAddr())) { 
  14.             return Response.status(400).entity("Missing ip address").build();
  15.         } else if (isBlank(info.getAppName())) { 
  16.             return Response.status(400).entity("Missing appName").build();
  17.         } else if (!appName.equals(info.getAppName())) { 
  18.             return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();
  19.         } else if (info.getDataCenterInfo() == null) { 
  20.             return Response.status(400).entity("Missing dataCenterInfo").build();
  21.         } else if (info.getDataCenterInfo().getName() == null) { 
  22.             return Response.status(400).entity("Missing dataCenterInfo Name").build();
  23.         }
  24.         //处理客户端可能在数据缺失的情况下向错误的DataCenterInfo注册的情况
  25.         // handle cases where clients may be registering with bad DataCenterInfo with missing data
  26.         DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
  27.         if (dataCenterInfo instanceof UniqueIdentifier) { 
  28.             String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
  29.             if (isBlank(dataCenterInfoId)) { 
  30.                 boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
  31.                 if (experimental) { 
  32.                     String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
  33.                     return Response.status(400).entity(entity).build();
  34.                 } else if (dataCenterInfo instanceof AmazonInfo) { 
  35.                     AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
  36.                     String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);
  37.                     if (effectiveId == null) { 
  38.                         amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());
  39.                     }
  40.                 } else { 
  41.                     logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
  42.                 }
  43.             }
  44.         }
  45.         //【重要】：这里在调用PeerAwareInstanceRegistry的register注册服务，使用的是实现类：InstanceRegistry
  46.         registry.register(info, "true".equals(isReplication));
  47.         return Response.status(204).build();  // 204 to be backwards compatible
  48.     }

ApplicationResource.addInstance看方法名就能推测出他是用来注册实例的方法，其中参数InstanceInfo 是客户端提交的注册信息，请求头中isReplicationd为true代表是从其他Eureka Server节点复制实例，如果是isReplication为false，代表是Eureka Client 注册的

在做了一些列参数判断之后，这里在调用PeerAwareInstanceRegistry的register注册服务，使用的是实现类：InstanceRegistry，这个类在之前有介绍过，就是Eureak Server用来实现服务注册，服务发现，服务续约，取消注册等的具体实现，他的继承关系如下：
在这里插入图片描述
跟踪下去，InstanceRegistry .register方法源码如下

  1. public class InstanceRegistry extends PeerAwareInstanceRegistryImpl implements ApplicationContextAware { 
  3.     public void register(final InstanceInfo info, final boolean isReplication) { 
  4.         //调用handleRegistration方法，抛出事件：EurekaInstanceRegisteredEvent
  5.         this.handleRegistration(info, this.resolveInstanceLeaseDuration(info), isReplication);
  6.         //调用父类PeerAwareInstanceRegistryImpl的register方法
  7.         super.register(info, isReplication);
  8.     }
  9.     //服务注册，抛出EurekaInstanceRegisteredEvent事件
  10.     private void handleRegistration(InstanceInfo info, int leaseDuration, boolean isReplication) { 
  11.         this.log("register " + info.getAppName() + ", vip " + info.getVIPAddress() + ", leaseDuration " + leaseDuration + ", isReplication " + isReplication);
  12.         this.publishEvent(new EurekaInstanceRegisteredEvent(this, info, leaseDuration, isReplication));
  13.     }
  14.   ...省略...  
  15.   }

这里在调用handleRegistration方法，抛出事件：EurekaInstanceRegisteredEvent后继续调用了super.register方法，即：PeerAwareInstanceRegistryImpl.register，继续跟踪下去：

  1. @Singleton
  2. public class PeerAwareInstanceRegistryImpl extends AbstractInstanceRegistry implements PeerAwareInstanceRegistry { 
  3.   /** 注册服务信息 InstanceInfo，并将此信息InstanceInfo复制到所有对等的eureka server节点。 如果这是来自其他副本节点的复制事件，则不会复制它。 * Registers the information about the {@link InstanceInfo} and replicates * this information to all peer eureka nodes. If this is replication event * from other replica nodes then it is not replicated. * * @param info * the {@link InstanceInfo} to be registered and replicated. * @param isReplication * true if this is a replication event from other replica nodes, * false otherwise. */
  4.     @Override
  5.     public void register(final InstanceInfo info, final boolean isReplication) { 
  6.         //租期失效时间 90 s
  7.         int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
  9.         if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) { 
  10.             //如果服务的租期失效时间大于默认的90s，则重新赋值租期时间
  11.             leaseDuration = info.getLeaseInfo().getDurationInSecs();
  12.         }
  13.         //服务注册
  14.         super.register(info, leaseDuration, isReplication);
  15.         //把注册的服务信息复制到其他的Eureka Server 节点，注意这里是Action.Register
  16.         replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
  17.     }
  18. }

该方法做了3个事情，

  • 1是更新租期失效时间，
  • 2是调用super.register服务注册(AbstractInstanceRegistry.register)
  • 3是调用replicateToPeers把服务实例拷贝到其他的Eureak Server节点

我们先看下super.register方法即：AbstractInstanceRegistry.register方法的源码

  1. /** * Registers a new instance with a given duration. * * @see com.netflix.eureka.lease.LeaseManager#register(java.lang.Object, int, boolean) */
  2.     public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) { 
  3.         try { 
  4.             //获取锁：ReentrantReadWriteLock.lock()
  5.             read.lock();
  6.             //根据注册的服务的名字取本地服务注册表中获取服务注册信息，如果该服务已经被注册了，那么registry中将会存在它
  7.             Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
  8.             REGISTER.increment(isReplication);
  9.             if (gMap == null) { 
  10.             //如果该服务实例没被注册，就把服务实例注册到本地的registry中，本质是一个Map
  11.                 final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
  12.                 //没有的话就添加到registry中
  13.                 gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
  14.                 if (gMap == null) { 
  15.                     gMap = gNewMap;
  16.                 }
  17.             }
  18.             //根据服务实例id获取服务的租约对象
  19.             Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
  20.             // Retain the last dirty timestamp without overwriting it, if there is already a lease
  21.             //如果已经有租约，则保留最后的脏时间戳而不覆盖它
  22.             if (existingLease != null && (existingLease.getHolder() != null)) { 
  23.                 //registry中已经存在的当前服务的最后修改时间的时间戳
  24.                 Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
  25.                 //提交注册的当前服务的最后修改时间
  26.                 Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
  27.                 logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
  28.             //因为如果时间戳相等，我们仍然采用远程传输的InstanceInfo而不是服务器本地副本
  29.                 // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
  30.                 // InstanceInfo instead of the server local copy.
  31.                 //如果已存在的该服务的修改时间 大于 当前提交注册的该服务的最后修改时间，
  32.                 //则采用registy中已存在的服务为准,因为要选择修改时间靠后的
  34.                 if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) { 
  35.                     logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
  36.                             " than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
  37.                     logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
  38.                     //使用现有的instanceInfo代替新的instanceInfo作为注册者
  39.                     registrant = existingLease.getHolder();
  40.                 }
  41.             } else { 
  42.                 //执行到这里，说明该服务是新注册
  43.                 // The lease does not exist and hence it is a new registration
  44.                 synchronized (lock) { 
  45.                 //这里在计算服务的续约频率值
  46.                     if (this.expectedNumberOfRenewsPerMin > 0) { 
  47.                         // Since the client wants to cancel it, reduce the threshold
  48.                         // (1
  49.                         // for 30 seconds, 2 for a minute)
  50.                         //(expectedNumberOfRenewsPerMin)期待的每分钟续订次数,默认是30s/个，给他增加到2，每分钟2个请求
  51.                         this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin + 2;
  52.                         //修改numberOfRenewsPerMinThreshold每分钟续约阀值 = 2 *（85%），
  53.                         //RenewalPercentThreshold是获取续订阈值百分比 
  54.                         this.numberOfRenewsPerMinThreshold =
  55.                                 (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
  56.                     }
  57.                 }
  58.                 logger.debug("No previous lease information found; it is new registration");
  59.             }
  60.             //创建租约对象，把注册实例和租期放进去
  61.             Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
  62.             if (existingLease != null) { 
  63.                 //设置服务上线时间
  64.                 lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
  65.             }
  66.             //以注册的实例的ID为key把服务实例存封装到 Map
  67.             gMap.put(registrant.getId(), lease);
  68.             //添加到注册队列
  69.             synchronized (recentRegisteredQueue) { 
  70.                 recentRegisteredQueue.add(new Pair<Long, String>(
  71.                         System.currentTimeMillis(),
  72.                         registrant.getAppName() + "(" + registrant.getId() + ")"));
  73.             }
  74.             // This is where the initial state transfer of overridden status happens
  75.             if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) { 
  76.                 logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
  77.                                 + "overrides", registrant.getOverriddenStatus(), registrant.getId());
  78.                 if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) { 
  79.                     logger.info("Not found overridden id {} and hence adding it", registrant.getId());
  80.                     //添加服务的OverriddenStatus
  81.                     overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
  82.                 }
  83.             }
  84.             InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
  85.             if (overriddenStatusFromMap != null) { 
  86.                 logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
  87.                 registrant.setOverriddenStatus(overriddenStatusFromMap);
  88.             }
  90.             //根据覆盖的状态规则设置状态
  91.             // Set the status based on the overridden status rules
  92.             InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
  93.             registrant.setStatusWithoutDirty(overriddenInstanceStatus);
  95.             // If the lease is registered with UP status, set lease service up timestamp
  96.             //如果租约已注册为UP状态，请设置租约服务启动时间戳记
  97.             if (InstanceStatus.UP.equals(registrant.getStatus())) { 
  98.                 //更新服务上线时间戳
  99.                 lease.serviceUp();
  100.             }
  101.             //服务动作：ADDED添加，MODIFIED修改，DELETED删除
  102.             registrant.setActionType(ActionType.ADDED);
  103.             //添加到最近更改的队列
  104.             recentlyChangedQueue.add(new RecentlyChangedItem(lease));
  105.             //最后更新时间
  106.             registrant.setLastUpdatedTimestamp();
  107.             //使当前应用的ResponseCache失效
  108.             invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
  109.             logger.info("Registered instance {}/{} with status {} (replication={})",
  110.                     registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
  111.         } finally { 
  112.             read.unlock();
  113.         }
  114.     }

总结一下，这个方法做了什么呢

  1. 判断当前服务是否已经被注册，如果是，则以最后更新时间为准，选择更新时间靠后的服务实例进行注册
  2. 维护实例的租约信息Lease，并放到Eureka Server本地维护维护的registry注册表中，本质是一个Map（ConcurrentHashMap>）
  3. 如果是服务是新注册的，把注册的实例封装成Leaset存储到registry注册表中，并更新每分钟续约阀值numberOfRenewsPerMinThreshold
  4. 维护了两个队列，recentRegisteredQueue最近注册队列，recentlyChangedQueue最近更改队列，这个队列可以用来获取最近操作的信息。
  5. 维护当前实例的OverriddenStatus
  6. 更新服务实例的最后更新时间戳
  7. 使ResponseCache缓存失效

该方法结束，我们回到PeerAwareInstanceRegistryImpl.register方法中，继续跟踪replicateToPeers 方法

  1. @Singleton
  2. public class PeerAwareInstanceRegistryImpl extends AbstractInstanceRegistry implements PeerAwareInstanceRegistry { 
  3.   /** * Replicates all eureka actions to peer eureka nodes except for replication * traffic to this node. * 将所有eureka操作复制到对等eureka节点 */
  4.     private void replicateToPeers(Action action, String appName, String id,
  5.                                   InstanceInfo info /* optional */,
  6.                                   InstanceStatus newStatus /* optional */, boolean isReplication) { 
  7.          //开始计时
  8.         Stopwatch tracer = action.getTimer().start();
  9.         try { 
  10.             //是否是其他节点复制过来的
  11.             if (isReplication) { 
  12.                 //最后一分钟的复制次数+1
  13.                 numberOfReplicationsLastMin.increment();
  14.             }
  15.             // If it is a replication already, do not replicate again as this will create a poison replication
  16.             //如果已经是复制，则不要再次复制
  17.             if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) { 
  18.                 return;
  19.             }
  20.             //遍历集群所有节点
  21.             for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) { 
  22.                 // If the url represents this host, do not replicate to yourself.
  23.                 //如果该URL代表此主机，请不要复制到您自己，当前节点不复制
  24.                 if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) { 
  25.                     continue;
  26.                 }
  27.                 //复制实例到其他某个Eureka
  28.                 replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
  29.             }
  30.         } finally { 
  31.             tracer.stop();
  32.         }
  33.     }

继续跟踪replicateInstanceActionsToPeers的源码

  1. /** * Replicates all instance changes to peer eureka nodes except for * replication traffic to this node. * */
  2.     private void replicateInstanceActionsToPeers(Action action, String appName,
  3.                                                  String id, InstanceInfo info, InstanceStatus newStatus,
  4.                                                  PeerEurekaNode node) { 
  5.         try { 
  6.             InstanceInfo infoFromRegistry = null;
  7.             CurrentRequestVersion.set(Version.V2);
  8.             //判断请求的是什么操作
  9.             switch (action) { 
  10.                 //取消注册，调用 PeerEurekaNode.cancel
  11.                 case Cancel:
  12.                     node.cancel(appName, id);
  13.                     break;
  14.                 //心跳请求，调用PeerEurekaNode.heartbeat
  15.                 case Heartbeat:
  16.                     InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);
  17.                     infoFromRegistry = getInstanceByAppAndId(appName, id, false);
  18.                     node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
  19.                     break;
  20.                 //服务注册调用PeerEurekaNode.register
  21.                 case Register:
  22.                     node.register(info);
  23.                     break;
  24.                  //状态修改调用PeerEurekaNode.statusUpdate
  25.                 case StatusUpdate:
  26.                     infoFromRegistry = getInstanceByAppAndId(appName, id, false);
  27.                     node.statusUpdate(appName, id, newStatus, infoFromRegistry);
  28.                     break;
  29.                   //状态删除调用PeerEurekaNode.deleteStatusOverride
  30.                 case DeleteStatusOverride:
  31.                     infoFromRegistry = getInstanceByAppAndId(appName, id, false);
  32.                     node.deleteStatusOverride(appName, id, infoFromRegistry);
  33.                     break;
  34.             }
  35.         } catch (Throwable t) { 
  36.             logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
  37.         }
  38.     }

这里在根据请求的动作类型选择PeerEurekaNode的不同方法，我们这里是服务注册.register，调用的是 PeerEurekaNode.register ，源码如下：

  1. /** * Sends the registration information of {@link InstanceInfo} receiving by * this node to the peer node represented by this class. * * @param info * the instance information {@link InstanceInfo} of any instance * that is send to this instance. * @throws Exception */
  2.     public void register(final InstanceInfo info) throws Exception { 
  3.         //到期时间当，前时间加上30s过期
  4.         long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
  5.         //封装InstanceReplicationTask 实例赋值任务到调度器中
  6.         batchingDispatcher.process(
  7.                 taskId("register", info),
  8.                 new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {                 
  9.                     public EurekaHttpResponse<Void> execute() { 
  10.                         //复制器客户端 HttpReplicationClient(JerseyReplicationClient)，执行注册
  11.                         //调用AbstractJerseyEurekaHttpClient#register
  12.                         return replicationClient.register(info);
  13.                     }
  14.                 },
  15.                 expiryTime
  16.         );
  17.     }

把InstanceInfo注册实例封装成InstanceReplicationTask实例复制任务，交给batchingDispatcher批量任务调度器去执行，replicationClient是HttpReplicationClient它的默认实现是JerseyEurekaHttpClient，底层会调用AbstractJerseyEurekaHttpClient#register的方法完成实例的注册，这里其实就和当时我们分析Eureka Client 服务注册的最后注册请求一样了

  1. public abstract class AbstractJerseyEurekaHttpClient implements EurekaHttpClient { 
  2.  public EurekaHttpResponse<Void> register(InstanceInfo info) { 
  3.          //请求地址
  4.         String urlPath = "apps/" + info.getAppName();
  5.         ClientResponse response = null;
  7.         EurekaHttpResponse var5;
  8.         try { 
  9.             Builder resourceBuilder = this.jerseyClient.resource(this.serviceUrl).path(urlPath).getRequestBuilder();
  10.             this.addExtraHeaders(resourceBuilder);
  11.             //把InstanceInfo作为参数，发送post请求提交服务注册
  12.             response = (ClientResponse)((Builder)((Builder)((Builder)resourceBuilder.header("Accept-Encoding", "gzip")).type(MediaType.APPLICATION_JSON_TYPE)).accept(new String[]{ "application/json"})).post(ClientResponse.class, info);
  13.             var5 = EurekaHttpResponse.anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
  14.         } finally { 
  15.             if (logger.isDebugEnabled()) { 
  16.                 logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", new Object[]{ this.serviceUrl, urlPath, info.getId(), response == null ? "N/A" : response.getStatus()});
  17.             }
  19.             if (response != null) { 
  20.                 response.close();
  21.             }
  23.         }
  25.         return var5;
  26.     }

到这里就结束了，在PeerEurekaNode中封装InstanceReplicationTask实例服务任务，通过EurekaHttpClient去发起请求(JerseyEurekaHttpClient)，最终通过JerseyEurekaHttpClient父类AbstractJerseyEurekaHttpClient#register方法注册，方法中把InstanceInfo作为参数使用POST提交请求。

总结

提示：这里对文章进行总结：例如：以上就是今天要讲的内容，本文仅仅简单介绍了pandas的使用，而pandas提供了大量能使我们快速便捷地处理数据的函数和方法。

发表评论

表情：
评论列表 （有 0 条评论，319人围观）

还没有评论，来说两句吧...

相关阅读