令牌桶算法和漏桶算法python_高并发系统限流-漏桶算法和令牌桶算法

参考：

http://www.cnblogs.com/LBSer/p/4083131.html

https://blog.csdn.net/scorpio3k/article/details/53103239

http://jinnianshilongnian.iteye.com/blog/2305117

http://iamzhongyong.iteye.com/blog/1742829

https://blog.csdn.net/qianshangding0708/article/details/104026669

https://blog.csdn.net/qianshangding0708/article/details/104026669

一、问题描述

某天A君突然发现自己的接口请求量突然涨到之前的10倍，没多久该接口几乎不可使用，并引发连锁反应导致整个系统崩溃。如何应对这种情况呢？生活给了我们答案：比如老式电闸都安装了保险丝，一旦有人使用超大功率的设备，保险丝就会烧断以保护各个电器不被强电流给烧坏。同理我们的接口也需要安装上“保险丝”，以防止非预期的请求对系统压力过大而引起的系统瘫痪，当流量过大时，可以采取拒绝或者引流等机制。

二、常用的限流算法

常用的限流算法有两种：漏桶算法和令牌桶算法。

漏桶算法思路很简单，水(请求)先进入到漏桶里，漏桶以一定的速度出水，当水流入速度过大会直接溢出，可以看出漏桶算法能强行限制数据的传输速率。

图1 漏桶算法示意图

对于很多应用场景来说，除了要求能够限制数据的平均传输速率外，还要求允许某种程度的突发传输。这时候漏桶算法可能就不合适了，令牌桶算法更为适合。如图2所示，令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。

图2 令牌桶算法示意图

三、限流工具类RateLimiter

Google开源工具包Guava提供了限流工具类RateLimiter，该类基于令牌桶算法来完成限流，非常易于使用。RateLimiter类的接口描述请参考：RateLimiter接口描述，具体使用请参考：RateLimiter使用实践。

下面是主要源码：

public double acquire() {

return acquire(1);

}

public double acquire(int permits) {

checkPermits(permits); //检查参数是否合法(是否大于0)

long microsToWait;

synchronized (mutex) { //应对并发情况需要同步

microsToWait = reserveNextTicket(permits, readSafeMicros()); //获得需要等待的时间

}

ticker.sleepMicrosUninterruptibly(microsToWait); //等待，当未达到限制时，microsToWait为0

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

}

private long reserveNextTicket(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.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;

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 com.netease.datastream.util.flowcontrol;

import java.io.BufferedWriter;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.OutputStreamWriter;

import java.util.Random;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.locks.ReentrantLock;

/**

*

* Created by inter12 on 15-3-18.

*

*/

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 tokenQueue = new ArrayBlockingQueue(

DEFAULT_BUCKET_SIZE);

private ScheduledExecutorService scheduledExecutorService = Executors

.newSingleThreadScheduledExecutor();

private volatile boolean isStart = false;

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.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,

// "please 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 false;

}

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(maxFlowRate);

}

// 初始化令牌生产者

TokenProducer tokenProducer = new TokenProducer(avgFlowRate, this);

scheduledExecutorService.scheduleAtFixedRate(tokenProducer, 0, 1,

TimeUnit.SECONDS);

isStart = true;