⾼并发系统限流-漏桶算法和令牌桶算法
⾼并发系统限流-漏桶算法和令牌桶算法
参考：
⼀、问题描述
某天A君突然发现⾃⼰的接⼝请求量突然涨到之前的10倍，没多久该接⼝⼏乎不可使⽤，并引发连锁反应导致整个系统崩溃。如何应对这种情况呢？⽣活给了我们答案：⽐如⽼式电闸都安装了保险丝，⼀旦有⼈使⽤超⼤功率的设备，保险丝就会烧断以保护各个电器不被强电流给烧坏。同理我们的接⼝也需要安装上“保险丝”，以防⽌⾮预期的请求对系统压⼒过⼤⽽引起的系统瘫痪，当流量过⼤时，可以采取拒绝或者引流等机制。
⼆、常⽤的限流算法
常⽤的限流算法有两种：漏桶算法和令牌桶算法。
漏桶算法思路很简单，⽔（请求）先进⼊到漏桶⾥，漏桶以⼀定的速度出⽔，当⽔流⼊速度过⼤会直接溢出，可以看出漏桶算法能强⾏限制数据的传输速率。
图1 漏桶算法⽰意图
对于很多应⽤场景来说，除了要求能够限制数据的平均传输速率外，还要求允许某种程度的突发传输。这时候漏桶算法可能就不合适了，令牌桶算法更为适合。如图2所⽰，令牌桶算法的原理是系统会以⼀个恒定的速度往桶⾥放⼊令牌，⽽如果请求需要被处理，则需要先从桶⾥获取⼀个令牌，当桶⾥没有令牌可取时，则拒绝服务。
图2 令牌桶算法⽰意图
三、限流⼯具类RateLimiter
Google开源⼯具包Guava提供了限流⼯具类RateLimiter，该类基于令牌桶算法来完成限流，⾮常易于使⽤。RateLimiter类的接⼝描述请参考：，具体使⽤请参考：。
下⾯是主要源码：
public double acquire() {
return acquire(1);
}
public double acquire(int permits) {
checkPermits(permits); //检查参数是否合法（是否⼤于0）
long microsToWait;
synchronized (mutex) { //应对并发情况需要同步
microsToWait = rerveNextTicket(permits, readSafeMicros()); //获得需要等待的时间
}
ticker.sleepMicrosUninterruptibly(microsToWait); //等待，当未达到限制时，microsToWait为0
染香
return 1.0 * microsToWait / Micros(1L);
}
private long rerveNextTicket(double requiredPermits, long nowMicros) {
resync(nowMicros); //补充令牌
long microsToNextFreeTicket = nextFreeTicketMicros - nowMicros;
double storedPermitsToSpend = Math.min(requiredPermits, this.storedPermits); //获取这次请求消耗的令牌数⽬
double freshPermits = requiredPermits - storedPermitsToSpend;
long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
+ (long) (freshPermits * stableIntervalMicros);
this.storedPermits -= storedPermitsToSpend; // 减去消耗的令牌
return microsToNextFreeTicket;
}
private void resync(long nowMicros) {
// if nextFreeTicket is in the past, resync to now
if (nowMicros > nextFreeTicketMicros) {
storedPermits = Math.min(maxPermits,
storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);
nextFreeTicketMicros = nowMicros;
}
}
服务治理---限流（令牌桶算法）
1、最近在写⼀个分布式服务的框架，对于分布式服务的框架来说，除了远程调⽤，还要进⾏服务的治理
当进⾏促销的时候，所有的资源都⽤来完成重要的业务，就⽐如双11的时候，主要的业务就是让⽤户查询商品，以及购买⽀付，
此时，⾦币查询、积分查询等业务就是次要的，因此要对这些服务进⾏服务的降级，典型的服务降级
算法是采⽤令牌桶算法，
因此在写框架的时候去研究了⼀下令牌桶算法
2、在实施QOS策略时，可以将⽤户的数据限制在特定的带宽，当⽤户的流量超过额定带宽时，超过的带宽将采取其它⽅式来处理。
要衡量流量是否超过额定的带宽，⽹络设备并不是采⽤单纯的数字加减法来决定的，也就是说，⽐如带宽为100K，⽽⽤户发来
的流量为110K，⽹络设备并不是靠110K减去100K等于10K，就认为⽤户超过流量10K。⽹络设备衡量流量是否超过额定带宽，
需要使⽤令牌桶来计算。下⾯详细介绍令牌桶算法机制：
当⽹络设备衡量流量是否超过额定带宽时，需要查看令牌桶，⽽令牌桶中会放置⼀定数量的令牌，⼀个令牌允许接⼝发送
或接收1bit数据（有时是1 Byte数据），当接⼝通过1bit数据后，同时也要从桶中移除⼀个令牌。当桶⾥没有令牌的时候，任何流
量都被视为超过额定带宽，只有当桶中有令牌时，数据才可以通过接⼝。令牌桶中的令牌不仅仅可以被移除，同样也可以往⾥添加，
所以为了保证接⼝随时有数据通过，就必须不停地往桶⾥加令牌，由此可见，往桶⾥加令牌的速度，就决定了数据通过接⼝的速度。
因此，我们通过控制往令牌桶⾥加令牌的速度从⽽控制⽤户流量的带宽。⽽设置的这个⽤户传输数据的速率被称为承诺信息速率（CIR），
通常以秒为单位。⽐如我们设置⽤户的带宽为1000 bit每秒，只要保证每秒钟往桶⾥添加1000个令牌即可。
3、举例：
将CIR设置为8000 bit/s，那么就必须每秒将8000个令牌放⼊桶中，当接⼝有数据通过时，就从桶中移除相应的令牌，每通过1 bit，
就从桶中移除1个令牌。当桶⾥没有令牌的时候，任何流量都被视为超出额定带宽，⽽超出的流量就要采取额外动作。每秒钟往桶⾥加的令牌
就决定了⽤户流量的速率，这个速率就是CIR，但是每秒钟需要往桶⾥加的令牌总数，并不是⼀次性加完的，⼀次性加进的令牌数量被称为Burst size（Bc），
如果Bc只是CIR的⼀半，那么很明显每秒钟就需要往桶⾥加两次令牌，每次加的数量总是Bc的数量。还有就是加令牌的时
间，Time interval（Tc），
Tc表⽰多久该往桶⾥加⼀次令牌，⽽这个时间并不能⼿⼯设置，因为这个时间可以靠CIR和Bc的关系计算得到， Bc/ CIR= Tc。
4、令牌桶算法图例
a. 按特定的速率向令牌桶投放令牌
b. 根据预设的匹配规则先对报⽂进⾏分类，不符合匹配规则的报⽂不需要经过令牌桶的处理，直接发送；
c. 符合匹配规则的报⽂，则需要令牌桶进⾏处理。当桶中有⾜够的令牌则报⽂可以被继续发送下去，同时令牌桶中的令牌 量按报⽂的长度做相应的减少；
d. 当令牌桶中的令牌不⾜时，报⽂将不能被发送，只有等到桶中⽣成了新的令牌，报⽂才可以发送。这就可以限制报⽂的流量只能是⼩于等于令牌⽣成的速度，达到限制流量的⽬的。
5、Java参考代码：
package comea.datastream.util.flowcontrol;
汉语言文学就业方向import java.io.BufferedWriter;
import java.io.FileOutputStream;
都护府
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Random;
import urrent.ArrayBlockingQueue;
import urrent.Executors;
import urrent.ScheduledExecutorService;
import urrent.TimeUnit;
import urrent.locks.ReentrantLock;
/**
* <pre>
* Created by inter12 on 15-3-18.
* </pre>
*/
public class TokenBucket {
// 默认桶⼤⼩个数即最⼤瞬间流量是64M
private static final int DEFAULT_BUCKET_SIZE = 1024 * 1024 * 64;
// ⼀个桶的单位是1字节形骸之外
private int everyTokenSize = 1;
// 瞬间最⼤流量
private int maxFlowRate;
// 平均流量
private int avgFlowRate;
// 队列来缓存桶数量：最⼤的流量峰值就是 = everyTokenSize*DEFAULT_BUCKET_SIZE 64M = 1 * 1024 *
// 1024 * 64
陶老师private ArrayBlockingQueue<Byte> tokenQueue = new ArrayBlockingQueue<Byte>(
金色的鱼钩故事DEFAULT_BUCKET_SIZE);
private ScheduledExecutorService scheduledExecutorService = Executors
.newSingleThreadScheduledExecutor();
private volatile boolean isStart = fal;
private ReentrantLock lock = new ReentrantLock(true);
private static final byte A_CHAR = 'a';
public TokenBucket() {
}
赵子琪演过的电视剧public TokenBucket(int maxFlowRate, int avgFlowRate) {
this.maxFlowRate = maxFlowRate;
this.maxFlowRate = maxFlowRate;
this.avgFlowRate = avgFlowRate;
}
public TokenBucket(int everyTokenSize, int maxFlowRate, int avgFlowRate) {
this.everyTokenSize = everyTokenSize;
this.maxFlowRate = maxFlowRate;
this.avgFlowRate = avgFlowRate;
}
public void addTokens(Integer tokenNum) {
// 若是桶已经满了，就不再家如新的令牌
for (int i = 0; i < tokenNum; i++) {
tokenQueue.offer(Byte.valueOf(A_CHAR));
}
}
老虎会爬树吗public TokenBucket build() {
start();
return this;
}
/**
* 获取⾜够的令牌个数
*
* @return
*/
public boolean getTokens(byte[] dataSize) {
// Preconditions.checkNotNull(dataSize);
// Preconditions.checkArgument(isStart,
// "plea invoke start method first !");
int needTokenNum = dataSize.length / everyTokenSize + 1;// 传输内容⼤⼩对应的桶个数
final ReentrantLock lock = this.lock;
lock.lock();
try {
boolean result = needTokenNum <= tokenQueue.size(); // 是否存在⾜够的桶数量
if (!result) {
return fal;
}
int tokenCount = 0;
for (int i = 0; i < needTokenNum; i++) {
Byte poll = tokenQueue.poll();
if (poll != null) {
tokenCount++;
}
}
return tokenCount == needTokenNum;
} finally {
lock.unlock();
}
}
public void start() {
// 初始化桶队列⼤⼩
if (maxFlowRate != 0) {
tokenQueue = new ArrayBlockingQueue<Byte>(maxFlowRate);
}
