注册中心 Eureka 源码解析 —— 基于令牌桶算法的 RateLimiter

摘要: 原创出处 http://www.iocoder.cn/Eureka/rate-limiter/ 「芋道源码」欢迎转载,保留摘要,谢谢!

本文主要基于 Eureka 1.8.X 版本

  • 1. 概述
  • 2. RateLimiter
  • 2.1 refillToken
  • 2.2 consumeToken
  • 3. RateLimitingFilter
  • 4. InstanceInfoReplicator
  • 666. 彩蛋

1. 概述

本文主要分享 RateLimiter 的代码实现和 RateLimiter 在 Eureka 中的应用

推荐 Spring Cloud 书籍

  • 请支持正版。下载盗版,等于主动编写低级 BUG
  • 程序猿DD —— 《Spring Cloud微服务实战》
  • 周立 —— 《Spring Cloud与Docker微服务架构实战》
  • 两书齐买,京东包邮。

推荐 Spring Cloud 视频

  • Java 微服务实践 - Spring Boot
  • Java 微服务实践 - Spring Cloud
  • Java 微服务实践 - Spring Boot / Spring Cloud

2. RateLimiter

com.netflix.discovery.util.RateLimiter ,基于Token Bucket Algorithm ( 令牌桶算法 )的速率限制器。

FROM 《接口限流实践》 令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌,而如果请求需要被处理,则需要先从桶里获取一个令牌,当桶里没有令牌可取时,则拒绝服务。

RateLimiter 目前支持分钟级秒级两种速率限制。构造方法如下:

public class RateLimiter {

    /**
     * 速率单位转换成毫秒
     */
    private final long rateToMsConversion;

    public RateLimiter(TimeUnit averageRateUnit) {
        switch (averageRateUnit) {
            case SECONDS: // 秒级
                rateToMsConversion = 1000;
                break;
            case MINUTES: // 分钟级
                rateToMsConversion = 60 * 1000;
                break;
            default:
                throw new IllegalArgumentException("TimeUnit of " + averageRateUnit + " is not supported");
        }
    }
}
  • averageRateUnit 参数,速率单位。构造方法里将 averageRateUnit 转换成 rateToMsConversion

调用 #acquire(...) 方法,获取令牌,并返回是否获取成功

// RateLimiter.java
/**
* 获取令牌( Token )
*
* @param burstSize 令牌桶上限
* @param averageRate 令牌再装平均速率
* @return 是否获取成功
*/
public boolean acquire(int burstSize, long averageRate) {
   return acquire(burstSize, averageRate, System.currentTimeMillis());
}

public boolean acquire(int burstSize, long averageRate, long currentTimeMillis) {
   if (burstSize <= 0 || averageRate <= 0) { // Instead of throwing exception, we just let all the traffic go
       return true;
   }

   // 填充 令牌
   refillToken(burstSize, averageRate, currentTimeMillis);
   // 消费 令牌
   return consumeToken(burstSize);
}
  • burstSize 参数 :令牌桶上限。
  • averageRate 参数 :令牌填充平均速率。
  • 我们举个 ? 来理解这两个参数 + 构造方法里的一个参数:
    • averageRateUnit = SECONDS
    • averageRate = 2000
    • burstSize = 10
    • 可获取 2000 个令牌。例如,每秒允许请求 2000 次。
    • 毫秒可填充 2000 / 1000 = 2消耗的令牌。
    • 毫秒可获取 10 个令牌。例如,每毫秒允许请求上限为 10 次,并且请求消耗掉的令牌,需要逐步填充。这里要注意下,虽然每毫秒允许请求上限为 10 次,这是在没有任何令牌被消耗的情况下,实际每秒允许请求依然是 2000 次。
    • 这就是基于令牌桶算法的限流的特点:让流量平稳,而不是瞬间流量。1000 QPS 相对平均的分摊在这一秒内,而不是第 1 ms 999 请求,后面 999 ms 0 请求
  • 从代码上看,#acquire(...) 分成两部分,我们分别解析,整体如下图:

2.1 refillToken

调用 #refillToken(...) 方法,填充已消耗的令牌。可能很多同学开始和我想的一样,一个后台每毫秒执行填充。为什么不适合这样呢?一方面,实际项目里每个接口都会有相应的 RateLimiter ,导致太多执行频率极高的后台任务;另一方面,获取令牌时才计算,多次令牌填充可以合并成一次,减少冗余和无效的计算。

代码如下:

  1: /**
  2:  * 速率单位转换成毫秒
  3:  */
  4: private final long rateToMsConversion;
  5: 
  6: /**
  7:  * 消耗令牌数
  8:  */
  9: private final AtomicInteger consumedTokens = new AtomicInteger();
 10: /**
 11:  * 最后填充令牌的时间
 12:  */
 13: private final AtomicLong lastRefillTime = new AtomicLong(0);
 14: 
 15: private void refillToken(int burstSize, long averageRate, long currentTimeMillis) {
 16:     // 获得 最后填充令牌的时间
 17:     long refillTime = lastRefillTime.get();
 18:     // 获得 过去多少毫秒
 19:     long timeDelta = currentTimeMillis - refillTime;
 20: 
 21:     // 计算 可填充最大令牌数量
 22:     long newTokens = timeDelta * averageRate / rateToMsConversion;
 23:     if (newTokens > 0) {
 24:         // 计算 新的填充令牌的时间
 25:         long newRefillTime = refillTime == 0
 26:                 ? currentTimeMillis
 27:                 : refillTime + newTokens * rateToMsConversion / averageRate;
 28:         // CAS 保证有且仅有一个线程进入填充
 29:         if (lastRefillTime.compareAndSet(refillTime, newRefillTime)) {
 30:             while (true) { // 死循环,直到成功
 31:                 // 计算 填充令牌后的已消耗令牌数量
 32:                 int currentLevel = consumedTokens.get();
 33:                 int adjustedLevel = Math.min(currentLevel, burstSize); // In case burstSize decreased
 34:                 int newLevel = (int) Math.max(0, adjustedLevel - newTokens);
 35:                 // CAS 避免和正在消费令牌的线程冲突
 36:                 if (consumedTokens.compareAndSet(currentLevel, newLevel)) {
 37:                     return;
 38:                 }
 39:             }
 40:         }
 41:     }
 42: }
  • 第 17 行 :获取最后填充令牌的时间( refillTime ) 。每次填充令牌,会设置 currentTimeMillisrefillTime
  • 第 19 行 :获得距离最后填充令牌的时间差( timeDelta ),用于计算需要填充的令牌数。
  • 第 22 行 :计算可填充的最大令牌数量( newTokens )。newTokens 可能超过 burstSize ,所以下面会有逻辑调整 newTokens
  • 第 25 至 27 行 :计算新的填充令牌的时间。为什么不能用 `currentTimeMillis` 呢?例如,averageRate = 500 &amp;&amp; averageRateUnit = SECONDS 时, 每 2 毫秒才填充一个令牌,如果设置 currentTimeMillis会导致不足以填充一个令牌的时长被吞了
  • 第 29 行 :通过 CAS 保证有且仅有一个线程进入填充逻辑。
  • 第 30 行 :死循环直到成功
  • 第 32 至 34 行 :计算新的填充令牌后的已消耗的令牌数量。
    • 第 33 行 :`burstSize` 可能调小,例如,系统接入分布式配置中心,可以远程调整该数值。如果此时 `burstSize` 更小,以它作为已消耗的令牌数量。
  • 第 36 行 :通过 CAS 保证避免覆盖设置正在消费令牌的线程。

2.2 consumeToken

#refillToken(...) 方法,填充消耗( 获取 )的令牌。

代码如下 :

  1: private boolean consumeToken(int burstSize) {
  2:     while (true) { // 死循环,直到没有令牌,或者获取令牌成功
  3:         // 没有令牌
  4:         int currentLevel = consumedTokens.get();
  5:         if (currentLevel >= burstSize) {
  6:             return false;
  7:         }
  8:         // CAS 避免和正在消费令牌或者填充令牌的线程冲突
  9:         if (consumedTokens.compareAndSet(currentLevel, currentLevel + 1)) {
 10:             return true;
 11:         }
 12:     }
 13: }
  • 第 2 行 :死循环直到没有令牌或者竞争获取令牌成功
  • 第 4 至 7 行 :没有令牌。
  • 第 9 至 11 行 :通过 CAS 避免和正在消费令牌或者填充令牌的线程冲突。

3. RateLimitingFilter

com.netflix.eureka.RateLimitingFilter ,Eureka-Server 限流过滤器。使用 RateLimiting ,保证 Eureka-Server 稳定性。

#doFilter(...) 方法,代码如下:

  1: @Override
  2: public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
  3:     // 获得 Target
  4:     Target target = getTarget(request);
  5: 
  6:     // Other Target ,不做限流
  7:     if (target == Target.Other) {
  8:         chain.doFilter(request, response);
  9:         return;
 10:     }
 11: 
 12:     HttpServletRequest httpRequest = (HttpServletRequest) request;
 13:     // 判断是否被限流
 14:     if (isRateLimited(httpRequest, target)) {
 15:         // TODO[0012]:监控相关,跳过
 16:         incrementStats(target);
 17:         // 如果开启限流,返回 503 状态码
 18:         if (serverConfig.isRateLimiterEnabled()) {
 19:             ((HttpServletResponse) response).setStatus(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
 20:             return;
 21:         }
 22:     }
 23:     chain.doFilter(request, response);
 24: }
  • 第 4 行 :调用 #getTarget() 方法,获取 Target。RateLimitingFilter 只对符合正在表达式 ^./apps(/[^/])?$ 的接口做限流,其中不包含 Eureka-Server 集群批量同步接口。
    • 点击 链接 查看 Target 枚举类代码。
    • 点击 链接 查看 #getTarget(…) 方法代码。
  • 第 14 行 :调用 #isRateLimited(...) 方法,判断是否被限流。代码如下: 1: private boolean isRateLimited(HttpServletRequest request, Target target) { 2: // 判断是否特权应用 3: if (isPrivileged(request)) { 4: logger.debug("Privileged {} request", target); 5: return false; 6: } 7: // 判断是否被超载( 限流 ) 8: if (isOverloaded(target)) { 9: logger.debug("Overloaded {} request; discarding it", target); 10: return true; 11: } 12: logger.debug("{} request admitted", target); 13: return false; 14: }
    • x
    • x
    • 第 3 至 6 行 :调用 #isPrivileged() 方法,判断是否为特权应用,对特权应用不开启限流逻辑。代码如下: private boolean isPrivileged(HttpServletRequest request) { // 是否对标准客户端开启限流 if (serverConfig.isRateLimiterThrottleStandardClients()) { return false; } // 以请求头( "DiscoveryIdentity-Name" ) 判断是否在标准客户端名集合内 Set<String> privilegedClients = serverConfig.getRateLimiterPrivilegedClients(); String clientName = request.getHeader(AbstractEurekaIdentity.AUTH_NAME_HEADER_KEY); return privilegedClients.contains(clientName) || DEFAULT_PRIVILEGED_CLIENTS.contains(clientName); }
    • 第 8 至 11 行 :调用 #isOverloaded(...) 方法,判断是否超载( 限流 )。代码如下: /** * Includes both full and delta fetches. */ private static final RateLimiter registryFetchRateLimiter = new RateLimiter(TimeUnit.SECONDS); /** * Only full registry fetches. */ private static final RateLimiter registryFullFetchRateLimiter = new RateLimiter(TimeUnit.SECONDS); private boolean isOverloaded(Target target) { int maxInWindow = serverConfig.getRateLimiterBurstSize(); // 10 int fetchWindowSize = serverConfig.getRateLimiterRegistryFetchAverageRate(); // 500 boolean overloaded = !registryFetchRateLimiter.acquire(maxInWindow, fetchWindowSize); if (target == Target.FullFetch) { int fullFetchWindowSize = serverConfig.getRateLimiterFullFetchAverageRate(); // 100 overloaded |= !registryFullFetchRateLimiter.acquire(maxInWindow, fullFetchWindowSize); } return overloaded; }
  • 第 18 至 21 行 :若 eureka.rateLimiter.enabled = true( 默认值 :false ,可配 ),返回 503 状态码。

4. InstanceInfoReplicator

com.netflix.discovery.InstanceInfoReplicator ,Eureka-Client 应用实例复制器。在 《Eureka 源码解析 —— 应用实例注册发现(一)之注册》「2.1 应用实例信息复制器」 有详细解析。

应用实例状态发生变化时,调用 #onDemandUpdate() 方法,向 Eureka-Server 发起注册,同步应用实例信息。InstanceInfoReplicator 使用 RateLimiter ,避免状态频繁发生变化,向 Eureka-Server 频繁同步。代码如下:

class InstanceInfoReplicator implements Runnable {

    /**
     * RateLimiter
     */
    private final RateLimiter rateLimiter;
    /**
     * 令牌桶上限,默认:2
     */
    private final int burstSize;
    /**
     * 令牌再装平均速率,默认:60 * 2 / 30 = 4
     */
    private final int allowedRatePerMinute;

    InstanceInfoReplicator(DiscoveryClient discoveryClient, InstanceInfo instanceInfo, int replicationIntervalSeconds, int burstSize) {
        // ... 省略其他代码

        this.rateLimiter = new RateLimiter(TimeUnit.MINUTES);
        this.replicationIntervalSeconds = replicationIntervalSeconds;
        this.burstSize = burstSize;

        this.allowedRatePerMinute = 60 * this.burstSize / this.replicationIntervalSeconds;
        logger.info("InstanceInfoReplicator onDemand update allowed rate per min is {}", allowedRatePerMinute);
    }

    public boolean onDemandUpdate() {
        if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) { // 限流
            scheduler.submit(new Runnable() {
                @Override
                public void run() {
                    logger.debug("Executing on-demand update of local InstanceInfo");
                    // 取消任务
                    Future latestPeriodic = scheduledPeriodicRef.get();
                    if (latestPeriodic != null && !latestPeriodic.isDone()) {
                        logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
                        latestPeriodic.cancel(false);
                    }
                    // 再次调用
                    InstanceInfoReplicator.this.run();
                }
            });
            return true;
        } else {
            logger.warn("Ignoring onDemand update due to rate limiter");
            return false;
        }
    }

}
  • #onDemandUpdate() 方法,调用 RateLimiter#acquire(…) 方法,获取令牌。
    • InstanceInfoReplicator 会固定周期检查本地应用实例是否有没向 Eureka-Server ,若未同步,则发起同步。在 《Eureka 源码解析 —— 应用实例注册发现(一)之注册》「2.1 应用实例信息复制器」 有详细解析。
    • Eureka-Client 向 Eureka-Server 心跳时,Eureka-Server 会对比应用实例的 `lastDirtyTimestamp` ,若 Eureka-Client 的更大,则 Eureka-Server 返回 404 状态码。Eureka-Client 接收到 404 状态码后,发起注册同步。在 Eureka 源码解析 —— 应用实例注册发现(二)之续租》「2.2 HeartbeatThread」 有详细解析。
    • 若获取成功,向 Eureka-Server 发起注册,同步应用实例信息。
    • 若获取失败,向 Eureka-Server 发起注册,同步应用实例信息。这样会不会有问题?答案是不会

原文发布于微信公众号 - 芋道源码(YunaiV)

原文发表时间:2018-08-13

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Android 研究

APK安装流程详解7——PackageManagerService的启动流程(上)

我们看到在SystemServer无参构造函数里面就是初始化mFactoryTestMode

1851
来自专栏JMCui

Netty 系列八(基于 WebSocket 的简单聊天室).

    之前写过一篇 Spring 集成 WebSocket 协议的文章 —— Spring消息之WebSocket ,所以对于 WebSocket 协议的介绍...

2825
来自专栏JetpropelledSnake

Django学习笔记之利用Form和Ajax实现注册功能

1705
来自专栏牛肉圆粉不加葱

[Spark源码剖析]Spark 延迟调度策略

在 Spark 中,若 task 与其输入数据在同一个 jvm 中,我们称 task 的本地性为 PROCESS_LOCAL,这种本地性(locality le...

1153
来自专栏喵了个咪的博客空间

[喵咪开源软件推荐(4)]Liunx跑分神器-unixbench

[喵咪开源软件推荐(4)]Liunx跑分神器-unixbench #w-blog博客 ? 哈喽大家好呀! 这次给大家带来一个Liunx跑分神奇,在笔者在老早之前...

3766
来自专栏老安的博客

Lotgstash日志切割示例

1533
来自专栏刘望舒

Android PMS处理APK的复制

在上一篇文章Android包管理机制之PackageInstaller安装APK中,我们学习了PackageInstaller是如何安装APK的,最后会将APK...

2145
来自专栏battcn

一起来学SpringBoot | 第二十二篇:轻松搞定重复提交(一)

在平时开发中,如果网速比较慢的情况下,用户提交表单后,发现服务器半天都没有响应,那么用户可能会以为是自己没有提交表单,就会再点击提交按钮重复提交表单,我们在开发...

3532
来自专栏Android 研究

APK安装流程详解12——PMS中的新安装流程上(拷贝)

从上面一片文章我们知道InstallAppProgress里面最后更新的代码是调用到PackageManager#installPackageWithVerif...

1411
来自专栏Kubernetes

Linux kernel Namespace源码分析

学习一下linux kernel namespace的代码还是很有必要的,让你对docker容器的namespace隔离有更深的认识。我的源码分析,是基于Lin...

9638

扫码关注云+社区

领取腾讯云代金券