11. DefaultController&RateLimiterController

127 阅读 0 评论 84 点赞

我是靠谱客的博主美好月饼，这篇文章主要介绍11. DefaultController&RateLimiterController，现在分享给大家，希望可以做个参考。

DefaultController

DefaultController 是默认使用的流量效果控制器，直接拒绝超出阈值的请求。当 QPS 超过限流规则配置的阈值，新的请求就会被立即拒绝，抛出 FlowException。适用于对系统处理能力明确知道的情况下，比如通过压测确定阈值。实际上我们很难测出这个阈值，因为一个服务可能部署在硬件配置不同的服务器上，并且随时都可能调整部署计划。

复制代码

public class DefaultController implements TrafficShapingController {
private static final int DEFAULT_AVG_USED_TOKENS = 0;
private double count;
private int grade;
public DefaultController(double count, int grade) {
this.count = count;
this.grade = grade;
}
@Override
public boolean canPass(Node node, int acquireCount) {
return canPass(node, acquireCount, false);
}
@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
//当前通过线程数或者node统计的qps数
int curCount = avgUsedTokens(node);
//当前数+请求数>阀值数，禁止通过(prioritized默认值为false)
if (curCount + acquireCount > count) {
if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {
long currentTime;
long waitInMs;
currentTime = TimeUtil.currentTimeMillis();
waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);
if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {
node.addWaitingRequest(currentTime + waitInMs, acquireCount);
node.addOccupiedPass(acquireCount);
sleep(waitInMs);
// PriorityWaitException indicates that the request will pass after waiting for {@link @waitInMs}.
throw new PriorityWaitException(waitInMs);
}
}
return false;
}
return true;
}
private int avgUsedTokens(Node node) {
if (node == null) {
return DEFAULT_AVG_USED_TOKENS;
}
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}
private void sleep(long timeMillis) {
try {
Thread.sleep(timeMillis);
} catch (InterruptedException e) {
// Ignore.
}
}
}

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
public class DefaultController implements TrafficShapingController {
private static final int DEFAULT_AVG_USED_TOKENS = 0;
private double count;
private int grade;
public DefaultController(double count, int grade) {
this.count = count;
this.grade = grade;
}
@Override
public boolean canPass(Node node, int acquireCount) {
return canPass(node, acquireCount, false);
}
@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
//当前通过线程数或者node统计的qps数
int curCount = avgUsedTokens(node);
//当前数+请求数>阀值数，禁止通过(prioritized默认值为false)
if (curCount + acquireCount > count) {
if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {
long currentTime;
long waitInMs;
currentTime = TimeUtil.currentTimeMillis();
waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);
if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {
node.addWaitingRequest(currentTime + waitInMs, acquireCount);
node.addOccupiedPass(acquireCount);
sleep(waitInMs);
// PriorityWaitException indicates that the request will pass after waiting for {@link @waitInMs}.
throw new PriorityWaitException(waitInMs);
}
}
return false;
}
return true;
}
private int avgUsedTokens(Node node) {
if (node == null) {
return DEFAULT_AVG_USED_TOKENS;
}
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}
private void sleep(long timeMillis) {
try {
Thread.sleep(timeMillis);
} catch (InterruptedException e) {
// Ignore.
}
}
}

RateLimiterController

Sentinel 匀速流控效果是漏桶算法结合虚拟队列等待机制实现的，可理解为存在一个虚拟的队列，请求在队列中排队通过，每（count/1000）毫秒可通过一个请求。虚拟队列的好处在于队列非真实存在，多核 CPU 多个请求并行通过时也可以通过，也就是说，实际通过的 QPS 会超过限流阈值的 QPS，但不会超很多。

要配置限流规则使用匀速通过效果控制器 RateLimiterController，则必须配置限流阈值类型为 GRADE_QPS，并且阈值要少于等于 1000。

此外匀速限流器可以使用于削峰填谷。场景：派单系统每隔3S搜集一定区域的订单和司机，然后调用最优匹配系统，适配出最优订单+司机组合。使用Elastic-job每隔3s触发一次，具体操作每个城市启动一个线程调用接口。100多个城市基本上同时调用最优化匹配系统，导致系统压力巨大。这里就可以考虑用匀速限流器实现，将这100个城市有序的匀速的分散，发生请求，有序也保证了每个城市都是间隔3s。

我们来看一下RateLimiterController 的构造方法

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
public class RateLimiterController implements TrafficShapingController {
//请求在虚拟队列中的最大等待时间，默认500毫秒
private final int maxQueueingTimeMs;
//限流QPS阀值
private final double count;
//最近一个请求通过的时间，用于计算下一个请求的预期通过时间
private final AtomicLong latestPassedTime = new AtomicLong(-1);
public RateLimiterController(int timeOut, double count) {
this.maxQueueingTimeMs = timeOut;
this.count = count;
}
}

RateLimiterController 实现的 canPass 方法源码如下

复制代码

@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
//....
// 假设1s之内限制200个请求,则间隔时间为5ms(count=200 acquireCount=1 costTime=5)
long currentTime = TimeUtil.currentTimeMillis();
long costTime = Math.round(1.0 * (acquireCount) / count * 1000);
// 当前请求预期通过的时间= 请求通过的间隔时间+最近一个请求通过的时间
long expectedTime = costTime + latestPassedTime.get();
// 预期时间<=当前时间，则放行
if (expectedTime <= currentTime) {
//存在并发问题，可能导致超过实际的限流阀值。但为性能考虑，误差可以忽略
latestPassedTime.set(currentTime);
return true;
} else {
//计算等待时间，超过最大等待时间则直接拒绝
long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
if (waitTime > maxQueueingTimeMs) {
return false;
} else {
try {
//重置latestPassedTime,二次校验
long oldTime = latestPassedTime.addAndGet(costTime);
waitTime = oldTime - TimeUtil.currentTimeMillis();
if (waitTime > maxQueueingTimeMs) {
//如果拒绝的话，回滚一个间隔
latestPassedTime.addAndGet(-costTime);
return false;
}
// 等待一段时间再放行
if (waitTime > 0) {
Thread.sleep(waitTime);
}
return true;
} catch (InterruptedException e) {
}
}
}
return false;
}

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

@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
//....
// 假设1s之内限制200个请求,则间隔时间为5ms(count=200 acquireCount=1 costTime=5)
long currentTime = TimeUtil.currentTimeMillis();
long costTime = Math.round(1.0 * (acquireCount) / count * 1000);
// 当前请求预期通过的时间= 请求通过的间隔时间+最近一个请求通过的时间
long expectedTime = costTime + latestPassedTime.get();
// 预期时间<=当前时间，则放行
if (expectedTime <= currentTime) {
//存在并发问题，可能导致超过实际的限流阀值。但为性能考虑，误差可以忽略
latestPassedTime.set(currentTime);
return true;
} else {
//计算等待时间，超过最大等待时间则直接拒绝
long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
if (waitTime > maxQueueingTimeMs) {
return false;
} else {
try {
//重置latestPassedTime,二次校验
long oldTime = latestPassedTime.addAndGet(costTime);
waitTime = oldTime - TimeUtil.currentTimeMillis();
if (waitTime > maxQueueingTimeMs) {
//如果拒绝的话，回滚一个间隔
latestPassedTime.addAndGet(-costTime);
return false;
}
// 等待一段时间再放行
if (waitTime > 0) {
Thread.sleep(waitTime);
}
return true;
} catch (InterruptedException e) {
}
}
}
return false;
}

long costTime = Math.round(1.0 * (acquireCount) / count * 1000); 这里要特别注意一下，匀速限流阀值(count)<=1000才有意义，如果count>1000&count<2000,则限流效果和1000是一样的。如果count>2000，则限流间隔costTime=0,实际上就是不限流。