令牌桶的实现_C语言实现栈

令牌桶的实现_C语言实现栈接上篇。Guava的令牌桶的实现中,包括一条设计哲学,需要大家注意:它允许瞬间的流量波峰超过QPS,但瞬间过后的请求将会等待较长的时间来缓解上次的波峰,以使得平均的QPS等于预定值。RateLimiter类提供了令牌桶的接口,它是一个抽象类,其子类有SmoothRateLimiter(也是个抽象类)以及孙子类SmoothBursty,SmoothWarmingUp。SmoothRateLimite…

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE使用 1年只要46元 售后保障 童叟无欺

接上篇。Guava的令牌桶的实现中,包括一条设计哲学,需要大家注意:它允许瞬间的流量波峰超过QPS,但瞬间过后的请求将会等待较长的时间来缓解上次的波峰,以使得平均的QPS等于预定值。

RateLimiter类提供了令牌桶的接口,它是一个抽象类,其子类有SmoothRateLimiter(也是个抽象类)以及孙子类SmoothBursty,SmoothWarmingUp。SmoothRateLimiter类实现了算法的核心部分,因次我们暂且只讨论SmoothRateLimiter和其实现类SmoothBursty。RateLimiter都是通过静态的create函数实例化。以create(double permitsPerSecond)为例。参数permitsPerSecond为配置的QPS。该方法简洁明了,屏蔽了很多用户无需关心的细节。

public static RateLimiter create(double permitsPerSecond) {

return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());

}

接着该方法调用了create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer())方法(该方法由于是包的访问权限,在实际的项目中,基本不会直接调用),同时创建了一个StopWatch,自动启动。

static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {

RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);

rateLimiter.setRate(permitsPerSecond);

return rateLimiter;

}

该方法创建了SmoothBursty实例,up-casting为RateLimiter。maxBurstSeconds固定为1,说明令牌桶中所能存储的的最大令牌数是1*QPS。接着调用setRate方法,该方法初始化一些重要的参数:

public final void setRate(double permitsPerSecond) {

checkArgument(

permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), “rate must be positive”);

synchronized (mutex()) {

doSetRate(permitsPerSecond, stopwatch.readMicros());

}

}

主要实现在SmoothRateLimiter中:

@Override

final void doSetRate(double permitsPerSecond, long nowMicros) {

resync(nowMicros);

double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;

this.stableIntervalMicros = stableIntervalMicros;

doSetRate(permitsPerSecond, stableIntervalMicros);

}

其中resync方法是一个关键的方法,在请求令牌时也会用到,后面还会说明:

void resync(long nowMicros) {

// if nextFreeTicket is in the past, resync to now

if (nowMicros > nextFreeTicketMicros) {

double newPermits = (nowMicros – nextFreeTicketMicros) / coolDownIntervalMicros();

storedPermits = min(maxPermits, storedPermits + newPermits);

nextFreeTicketMicros = nowMicros;

}

}

从中可以看出,如果nowMicros大于nextFreeTicketMicros,会重新计算nextFreeTicketMicros和storedPermit的值。设置stableIntervalMicros,该字段表示1/QPS,即生产令牌的速率。

接着调用doSetRate方法,该方法在SmoothBursty类中。

@Override

void doSetRate(double permitsPerSecond, double stableIntervalMicros) {

double oldMaxPermits = this.maxPermits;

maxPermits = maxBurstSeconds * permitsPerSecond;

if (oldMaxPermits == Double.POSITIVE_INFINITY) {

// if we don’t special-case this, we would get storedPermits == NaN, below

storedPermits = maxPermits;

} else {

storedPermits =

(oldMaxPermits == 0.0)

? 0.0 // initial state

: storedPermits * maxPermits / oldMaxPermits;

}

}

初始化maxPermits和storePermits,后者永远不会大于前者。

到此,rateLimiter初始化完成。除了resync方法,在不重新设置rate的情况,其他方法不在处理请求时用到,暂时忽略。

下面看关键的令牌申请的过程。

首先调用acquire()方法,申请令牌,无参数表示申请一个。

public double acquire() {

return acquire(1);

}

接着调用acquire(int permits)方法:

@CanIgnoreReturnValue

public double acquire(int permits) {

long microsToWait = reserve(permits);

stopwatch.sleepMicrosUninterruptibly(microsToWait);

return 1.0 * microsToWait / SECONDS.toMicros(1L);

}

reserve方法返回获取令牌所需要等待的时间,stopwatch阻塞当前线程,最后返回线程休眠的秒数。如果microsToWait为0,表示立即返回。

final long reserve(int permits) {

checkPermits(permits);

synchronized (mutex()) {

return reserveAndGetWaitLength(permits, stopwatch.readMicros());

}

}

reserve需要获取锁才可以操作,这也是令牌桶线程安全的原因,以下操作都在同步代码块中。

继续reserveAndGetWaitLength方法。

final long reserveAndGetWaitLength(int permits, long nowMicros) {

long momentAvailable = reserveEarliestAvailable(permits, nowMicros);

return max(momentAvailable – nowMicros, 0);

}

首先调用reserveEarliestAvailable,方法名说明了返回值的意义:即返回满足当前请求的最早的时钟,该值大于等于nowMicros。如何保证这一点的呢?我们看该方法:

@Override

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {

resync(nowMicros);

long returnValue = nextFreeTicketMicros;

double storedPermitsToSpend = min(requiredPermits, this.storedPermits);

double freshPermits = requiredPermits – storedPermitsToSpend;

long waitMicros =

storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)

+ (long) (freshPermits * stableIntervalMicros);

this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);

this.storedPermits -= storedPermitsToSpend;

return returnValue;

}

这十多行代码是整个算法实现的核心,重点说明:

首先调用resync(nowMicros),重置nextFreeTicketMicros。如果nowMicros在nextFreeTicketMicros之后,nextFreeTicketMicros=nowMicros,并往storedPermits中增加这段时间能产生的令牌。

返回值设置为当前的nextFreeTicketMicros。为什么要这样设置呢?因为如果nowMicros大于nextFreeTicketMicros,说明令牌桶肯定能满足需求(无论请求的令牌数目是多少,参见最上面的设计哲学),而resync方法已经修改了nextFreeTicketMicros值为nowMicros值,逐层返回给调用者时,等待时间为0,线程无需等待;反之,如果nowMicros小于等于nextFreeTicketMicros,说明请求过快,线程需要等待,等待的时间就是nextFreeTicketMicros-nowMicros。

接下来,storedPermitsToSpend代表令牌桶中已有的令牌数,可以用于当前的请求。但未必满足需求。

其次,freshPermits代表需要新生成的令牌数。如果storedPermits已经满足需求,则freshPermits为0。

再次,计算新生成令牌需要花费的时间,这些需要后来者偿还。

然后修改nextFreeTicketMicros的值。

最后修改storedPermits值。

至此整个处理过程结束。

经过上面的代码梳理,详细大家对RateLimiter的代码有个比较清晰的认识,但要加深理解,还需要多做debug和test。

Guava包里面包括了很多test case。我们可以把test类单拿出来,根据自己的情况添加相应的case即可。该类是com.google.common.util.concurrent. RateLimiterTest。由于很多类都使用了默认访问权限,我们需要定义一个 com.google.common.util.concurrent包,导入RateLimiterTest类。该类中,guava提供了一个FakeStopwatch的nested class。它能够让时钟按照我们的要求暂停,休眠随意的时长,并记录休眠和请求对应的事件,并已特定的格式输出。例如:R1.00代表请求给定的令牌延迟了1秒;U1.05表示stopwatch休眠1.05秒,即模拟时钟过了1.05秒。例如一个测试通过的case:

public void testSimple() {

RateLimiter limiter = RateLimiter.create(5.0, stopwatch);

limiter.acquire(); // R0.00

limiter.acquire(); // R0.20

limiter.acquire(); // R0.20

stopwatch.sleepMillis(1000); // U1.00

assertEvents(“R0.00”, “R0.20”, “R0.20”, “U1.00”);

}

下面提供一个case,验证下大家的理解。

public void testOneSecondBurst3() {

RateLimiter limiter = RateLimiter.create(1.0, stopwatch);

limiter.acquire(1); // R值?

stopwatch.sleepMillis(1050);//U值?

limiter.acquire(1); // R值? nowMicros? nextFree?

stopwatch.sleepMillis(950);

limiter.acquire(1); // R值? nowMicros? nextFree?

stopwatch.sleepMillis(1000);

limiter.acquire(1); // R值? nowMicros? nextFree?

}

关注公众号“码农走向艺术”,回复消息可以获取答案幺:)

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/179115.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号