【算法】服务治理---限流(令牌桶算法)
576 ✦ 843842144@qq.com

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;


/**
* <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 = 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<Byte>(maxFlowRate);
}

// 初始化令牌生产者
TokenProducer tokenProducer = new TokenProducer(avgFlowRate, this);
scheduledExecutorService.scheduleAtFixedRate(tokenProducer, 0, 1,
TimeUnit.SECONDS);
isStart = true;

}

public void stop() {
isStart = false;
scheduledExecutorService.shutdown();
}

public boolean isStarted() {
return isStart;
}

class TokenProducer implements Runnable {

private int avgFlowRate;
private TokenBucket tokenBucket;

public TokenProducer(int avgFlowRate, TokenBucket tokenBucket) {
this.avgFlowRate = avgFlowRate;
this.tokenBucket = tokenBucket;
}

@Override
public void run() {
tokenBucket.addTokens(avgFlowRate);
}
}

public static TokenBucket newBuilder() {
return new TokenBucket();
}

public TokenBucket everyTokenSize(int everyTokenSize) {
this.everyTokenSize = everyTokenSize;
return this;
}

public TokenBucket maxFlowRate(int maxFlowRate) {
this.maxFlowRate = maxFlowRate;
return this;
}

public TokenBucket avgFlowRate(int avgFlowRate) {
this.avgFlowRate = avgFlowRate;
return this;
}

private String stringCopy(String data, int copyNum) {

StringBuilder sbuilder = new StringBuilder(data.length() * copyNum);

for (int i = 0; i < copyNum; i++) {
sbuilder.append(data);
}

return sbuilder.toString();

}

public static void main(String[] args) throws IOException,
InterruptedException {

tokenTest();
}

private static void arrayTest() {
ArrayBlockingQueue<Integer> tokenQueue = new ArrayBlockingQueue<Integer>(
10);
tokenQueue.offer(1);
tokenQueue.offer(1);
tokenQueue.offer(1);
System.out.println(tokenQueue.size());
System.out.println(tokenQueue.remainingCapacity());
}

private static void tokenTest() throws InterruptedException, IOException {
TokenBucket tokenBucket = TokenBucket.newBuilder().avgFlowRate(512)
.maxFlowRate(1024).build();

BufferedWriter bufferedWriter = new BufferedWriter(
new OutputStreamWriter(new FileOutputStream("D:/ds_test")));
String data = "xxxx";// 四个字节
for (int i = 1; i <= 1000; i++) {

Random random = new Random();
int i1 = random.nextInt(100);
boolean tokens = tokenBucket.getTokens(tokenBucket.stringCopy(data,
i1).getBytes());
TimeUnit.MILLISECONDS.sleep(100);
if (tokens) {
bufferedWriter.write("token pass --- index:" + i1);
System.out.println("token pass --- index:" + i1);
} else {
bufferedWriter.write("token rejuect --- index" + i1);
System.out.println("token rejuect --- index" + i1);
}

bufferedWriter.newLine();
bufferedWriter.flush();
}

bufferedWriter.close();
}

}


Submit
no comments yet~