Eureka源码—EurekaServer

发表于 | 分类于 ， | 评论数：   |   字数统计: 2,383   |   阅读时长: 11min | 热度： ℃

前言

本篇文章主要解读Eureka-Server初始化的相关源码，文章较长，请谨慎点击！

EurekaServerConfig

EurekaServerConfig简介

所属包：com.netflix.eureka.EurekaSeverConfig，Eureka-Server配置接口。

EurekaServerConfig源码

博主已添加中文注释，点击查看。

EurekaServerConfig重要属性或者重要方法

  • 自我保护相关：
    • shouldEnableSelfPreservation()：是否开启自我保护机制，默认为true。
    • getRenewalPercentThreshold()：自我保护百分比阈值，超过这个值就会开启自我保护模式。默认为0.85。
    • getRenewalThresholdUpdateIntervalMs（）:自我保护阈值更新的时间间隔，单位为毫秒（ms）。默认为 15 * 60 * 1000 ms。
  • 注册信息相关：
    • getWaitTimeInMsWhenSyncEmpty()：如果eureka在启动的时候，获取不到注册信息，就等待。单位为毫秒（ms）。默认为 1000 * 60 * 5 ms。
    • getPeerNodeConnectTimeoutMs()：eureka集群节点之间请求连接超时的时间（进行复制信息操作），单位为毫秒（ms）。默认为 1000 ms。
    • getPeerNodeReadTimeoutMs()：eureka集群节点之间请求读取超时的时间（进行复制信息操作），单位为毫秒（ms），默认为 5000 ms。
    • getRetentionTimeInMSInDeltaQueue()：获取增量信息缓存的时间，以避免客户端检索的时候丢失，单位为毫秒。默认为 30 * 60 * 1000 ms。
    • getDeltaRetentionTimerIntervalInMs()：获取清理过期的增量信息任务应唤醒的时间间隔，单位为毫秒。默认为 30 * 1000 ms。
    • shouldDisableDelta()：是否可以将增量信息提供给客户端，默认为false。
    • shouldSyncWhenTimestampDiffers()：否同步应用实例信息，当应用实例信息最后更新时间戳( lastDirtyTimestamp )发生改变。默认为true。
    • getRegistrySyncRetries()：获取eureka服务器启动时，从远程 Eureka-Server 读取失败重试次数。默认为5。
    • getRegistrySyncRetryWaitMs()：Eureka-Server 启动时，从远程 Eureka-Server 读取注册信息失败，再次尝试获取的等待( sleep )间隔，单位为毫秒，默认为30 * 1000 ms。
    • shouldDisableDeltaForRemoteRegions()：是否将增量信息提供给客户端或远程区域，默认为false。
    • getRemoteRegionConnectTimeoutMs()：获取连接到远程Eureka-Server的超时时间，单位为毫秒，默认为2000ms。
    • getRemoteRegionReadTimeoutMs()：获取读取远程Eureka-Server的超时时间，单位为毫秒，默认为5000ms。
    • getRemoteRegionUrlsWithName()：获取远程Eureka-Server的地址，key为Eureka-server名，value为Eureka-Server地址。
    • getRemoteRegionAppWhitelist(@Nullable String regionName)：获取远程Eureka-Server获取注册的Applications集合。
    • getRemoteRegionRegistryFetchInterval()：获取从远程Eureka-Server获取注册信息的时间间隔，单位为秒，默认为30s。
  • 限流相关：
    • isRateLimiterEnabled()：是否开启请求限流，默认为false。
    • isRateLimiterThrottleStandardClients()：是否限制非标准客户端的访问，默认为false。标准客户端通过请求头的 DiscoveryIdentity-Name 来判断，是否在标准客户端名集合里。
    • getRateLimiterPrivilegedClients()：获取标准客户端名集合。标准客户端名集合。默认包含DefaultClient 和 DefaultServer 。
    • getRateLimiterBurstSize()：速率限制的burst size，默认为10。令牌桶算法。
    • getRateLimiterRegistryFetchAverageRate()：指定增量获取注册信息的平均执行注册请求速率，默认为500。
    • getRateLimiterFullFetchAverageRate()：指定全量获取注册信息的平均执行注册请求速率，默认为100。
  • getEvictionIntervalTimerInMs()：租约过期定时任务执行频率，单位为毫秒。默认为 60 * 1000 ms。

DefaultEurekaServerConfig

DefaultEurekaServerConfig

所属包：com.netflic.eureka.DefaultEurekaServerConfig，基于配置文件的Eureka-Server配置实现类。

DefaultEurekaServerConfig源码

博主已添加中文注释，点击查看。

DefaultEurekaServerConfig重要属性或者重要方法

  • 其他实现方法基本类似，都是从配置文件中获取值，举一个例子：
    • shouldEnableSelfPreservation()：是否开启自我保护模式，并传一个默认值（这里是true）。

EurekaBootStrap

EurekaBootStrap类结构图

EurekaBootStrap简介

所属包：com.netflix.eureka.EurekaBootStrap，Eureka-Server启动入口。EurekaBootStrap 实现了 javax.servlet.ServletContextListener 接口，在 Servlet 容器( 例如 Tomcat、Jetty )启动时，调用 contextInitialized() 方法，初始化 Eureka-Server 。源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public class EurekaBootStrap implements ServletContextListener {
    // 省略部分代码
    @Override
    public void contextInitialized(ServletContextEvent event) {
        try {
            // 初始化Eureka配置环境
            initEurekaEnvironment();
            // 初始化Eureka-Server上下文
            initEurekaServerContext();

            ServletContext sc = event.getServletContext();
            sc.setAttribute(EurekaServerContext.class.getName(), serverContext);
        } catch (Throwable e) {
            logger.error("Cannot bootstrap eureka server :", e);
            throw new RuntimeException("Cannot bootstrap eureka server :", e);
        }
    }
    // 省略部分代码
}

EurekaBootStrap源码

博主已添加中文注释，点击查看。

EurekaBootStrap重要属性或者重要方法

  • Eureka-Server上下文相关：

    • com.netflix.eureka.EurekaServerContext：Eureka-Server上下文接口，包含了初始化，关闭，获取EurekaServer配置对象，获取集群节点集合，获取编解码器，获取应用实例注册表，获取应用信息管理器这些方法。源码如下：

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      public interface EurekaServerContext {

          void initialize() throws Exception;

          void shutdown() throws Exception;

          EurekaServerConfig getServerConfig();

          PeerEurekaNodes getPeerEurekaNodes();

          ServerCodecs getServerCodecs();

          PeerAwareInstanceRegistry getRegistry();

          ApplicationInfoManager getApplicationInfoManager();

      }

    • com.netflix.eureka.DefaultEurekaServerContext：Eureka-Server上下文默认实现类，源码如下：

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      46
      47
      48
      49
      50
      51
      52
      53
      54
      55
      56
      57
      58
      59
      60
      61
      62
      63
      64
      65
      66
      67
      68
      69
      70
      71
      72
      73
      74
      75
      76
      77
      78
      79
      80
      81
      82
      83
      84
      85
      86
      @Singleton
      public class DefaultEurekaServerContext implements EurekaServerContext {
          private static final Logger logger = LoggerFactory.getLogger(DefaultEurekaServerContext.class);

          /**
           * Eureka-Server配置对象
           */
          private final EurekaServerConfig serverConfig;
          /**
           * 编解码器
           */
          private final ServerCodecs serverCodecs;
          /**
           * 应用实例信息注册表
           */
          private final PeerAwareInstanceRegistry registry;
          /**
           * Eureka-Server集群节点集合
           */
          private final PeerEurekaNodes peerEurekaNodes;
          /**
           * 应用信息管理器
           */
          private final ApplicationInfoManager applicationInfoManager;

          @Inject
          public DefaultEurekaServerContext(EurekaServerConfig serverConfig,
                                     ServerCodecs serverCodecs,
                                     PeerAwareInstanceRegistry registry,
                                     PeerEurekaNodes peerEurekaNodes,
                                     ApplicationInfoManager applicationInfoManager) {
              this.serverConfig = serverConfig;
              this.serverCodecs = serverCodecs;
              this.registry = registry;
              this.peerEurekaNodes = peerEurekaNodes;
              this.applicationInfoManager = applicationInfoManager;
          }

          @PostConstruct
          @Override
          public void initialize() throws Exception {
              logger.info("Initializing ...");
              // 启动Eureka-Server集群节点集合
              peerEurekaNodes.start();
              // 初始化应用实例信息注册表
              registry.init(peerEurekaNodes);
              logger.info("Initialized");
          }

          @PreDestroy
          @Override
          public void shutdown() throws Exception {
              logger.info("Shutting down ...");
              // 关闭应用实例信息注册表
              registry.shutdown();
              // 关闭Eureka-Server集群节点集合
              peerEurekaNodes.shutdown();
              logger.info("Shut down");
          }

          @Override
          public EurekaServerConfig getServerConfig() {
              return serverConfig;
          }

          @Override
          public PeerEurekaNodes getPeerEurekaNodes() {
              return peerEurekaNodes;
          }

          @Override
          public ServerCodecs getServerCodecs() {
              return serverCodecs;
          }

          @Override
          public PeerAwareInstanceRegistry getRegistry() {
              return registry;
          }

          @Override
          public ApplicationInfoManager getApplicationInfoManager() {
              return applicationInfoManager;
          }

      }

    • com.netflix.eureka.EurekaServerContextHolder：Eureka-Server上下文持有者，可以通过它获取上下文，源码如下：

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      public class EurekaServerContextHolder {

          /**
           * Eureka-Server上下文
           */
          private final EurekaServerContext serverContext;

          private EurekaServerContextHolder(EurekaServerContext serverContext) {
              this.serverContext = serverContext;
          }

          /**
           * 获取Eureka-Server上下文
           *
           * @return Eureka-Server上下文
           */
          public EurekaServerContext getServerContext() {
              return this.serverContext;
          }

          /**
           * Eureka-Server上下文持有者
           */
          private static EurekaServerContextHolder holder;

          /**
           * 初始化
           *
           * @param serverContext Eureka-Server上下文
           */
          public static synchronized void initialize(EurekaServerContext serverContext) {
              holder = new EurekaServerContextHolder(serverContext);
          }

          /**
           * 获取Eureka-Server上下文持有者
           *
           * @return Eureka-Server上下文持有者
           */
          public static EurekaServerContextHolder getInstance() {
              return holder;
          }
      }

StatusFilter

StatusFilter简介

所属包：com.netflix.eureka.StatusFilter，根据InstanceStatus过滤Eureka-Server。当 Eureka-Server 未处于开启( InstanceStatus.UP )状态，返回 HTTP 状态码 307 重定向 。

StatusFilter源码

1
2
3
4
5
6
7
8
9
10
11
12
public void doFilter(ServletRequest request, ServletResponse response,
                         FilterChain chain) throws IOException, ServletException {
    InstanceInfo myInfo = ApplicationInfoManager.getInstance().getInfo();
    InstanceStatus status = myInfo.getStatus();
    if (status != InstanceStatus.UP && response instanceof HttpServletResponse) {
        HttpServletResponse httpRespone = (HttpServletResponse) response;
        httpRespone.sendError(SC_TEMPORARY_REDIRECT,
                              "Current node is currently not ready to serve requests -- current status: "
                              + status + " - try another DS node: ");
    }
    chain.doFilter(request, response);
}

ServerRequestAuthFilter

ServerRequestAuthFilter简介

所属包：com.netflix.eureka.ServerRequestAuthFilter，用于客户端请求的身份验证过滤器。目前，它只记录来自标题信息的支持的客户端标识数据。

ServerRequestAuthFilter源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
protected void logAuth(ServletRequest request) {
    if (serverConfig.shouldLogIdentityHeaders()) {
        if (request instanceof HttpServletRequest) {
            HttpServletRequest httpRequest = (HttpServletRequest) request;

            String clientName = getHeader(httpRequest, AbstractEurekaIdentity.AUTH_NAME_HEADER_KEY);
            String clientVersion = getHeader(httpRequest, AbstractEurekaIdentity.AUTH_VERSION_HEADER_KEY);

            DynamicCounter.increment(MonitorConfig.builder(NAME_PREFIX + clientName + "-" + clientVersion).build());
        }
    }
}

protected String getHeader(HttpServletRequest request, String headerKey) {
    String value = request.getHeader(headerKey);
    return Strings.isNullOrEmpty(value) ? UNKNOWN : value;
}

RateLimitingFilter

RateLimitingFilter简介

所属包：com.netflix.eureka.RateLimitingFilter，请求速率限制过滤器。

GzipEncodingEnforcingFilter

GzipEncodingEnforcingFilter简介

所属包：com.netflix.eureka.GzipEncodingEnforcingFilter，gzip编码。最初Eureka仅支持非压缩响应, 对于大型注册管理机构来说，它非常低效，因此增加了gzip编码。如今，所有现代HTTP客户端都透明地支持gzip HTTP响应，因此不再需要维护未压缩的内容。 通过添加此过滤器，Eureka 服务器将只接受明确支持gzip编码回复的GET请求。, 在即将到来的次要版本中，非压缩回复将完全丢弃，因此此过滤器将成为必需。

GzipEncodingEnforcingFilter源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
    HttpServletRequest httpRequest = (HttpServletRequest) request;
    if ("GET".equals(httpRequest.getMethod())) {
        String acceptEncoding = httpRequest.getHeader(HttpHeaders.ACCEPT_ENCODING);
        if (acceptEncoding == null) {
            chain.doFilter(addGzipAcceptEncoding(httpRequest), response);
            return;
        }
        if (!acceptEncoding.contains("gzip")) {
            ((HttpServletResponse) response).setStatus(HttpServletResponse.SC_NOT_ACCEPTABLE);
            return;
        }
    }
    chain.doFilter(request, response);
}
private static HttpServletRequest addGzipAcceptEncoding(HttpServletRequest request) {
    return new HttpServletRequestWrapper(request) {

        @Override
        public Enumeration<String> getHeaders(String name) {
            if (HttpHeaders.ACCEPT_ENCODING.equals(name)) {
                return new EnumWrapper<String>("gzip");
            }
            return new EnumWrapper<String>(super.getHeaders(name), HttpHeaders.ACCEPT_ENCODING);
        }

        @Override
        public Enumeration<String> getHeaderNames() {
            return new EnumWrapper<String>(super.getHeaderNames(), HttpHeaders.ACCEPT_ENCODING);
        }

        @Override
        public String getHeader(String name) {
            if (HttpHeaders.ACCEPT_ENCODING.equals(name)) {
                return "gzip";
            }
            return super.getHeader(name);
        }
    };
}

private static class EnumWrapper<E> implements Enumeration<E> {

    private final Enumeration<E> delegate;
    private final AtomicReference<E> extraElementRef;

    private EnumWrapper(E extraElement) {
        this(null, extraElement);
    }

    private EnumWrapper(Enumeration<E> delegate, E extraElement) {
        this.delegate = delegate;
        this.extraElementRef = new AtomicReference<>(extraElement);
    }

    @Override
    public boolean hasMoreElements() {
        return extraElementRef.get() != null || delegate != null && delegate.hasMoreElements();
    }

    @Override
    public E nextElement() {
        E extra = extraElementRef.getAndSet(null);
        if (extra != null) {
            return extra;
        }
        if (delegate == null) {
            throw new NoSuchElementException();
        }
        return delegate.nextElement();
    }
}

ServletContainer

ServletContainer简介

所属包：com.sun.jersey.spi.container.servlet.ServletContainer，Jersey MVC 请求过滤器。

总结

我们可以看到Eureka-Server的生成，是EurekaBootStrap实现了 javax.servlet.ServletContextListener 接口，在 Servlet 容器( 例如 Tomcat、Jetty )启动时，调用 contextInitialized() 方法初始化Eureka-Server。Eureka-Server内嵌 Eureka-Client，用于和Eureka-Server集群里其他节点通信交互。

欢迎关注博主其他的文章。

感谢您的支持！

本文标题:Eureka源码—EurekaServer

文章作者:yoga

发布时间:2018年03月23日 - 20:03

原始链接:https://yoga0521.github.io/2018/03/23/Eureka源码—EurekaServer/

版权声明:本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。 转载请注明出处！