令牌桶(Token Bucket)

概要

　　限流可以认为服务降级的一种，限流就是限制系统的输入和输出流量已达到保护系统的目的。一般来说系统的吞吐量是可以被测算的，为了保证系统的稳定运行，一旦达到的需要限制的阈值，就需要限制流量并采取一些措施以完成限制流量的目的。比如：延迟处理，拒绝处理，或者部分拒绝处理等等。

令牌桶算法

　　令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。 当桶满时，新添加的令牌被丢弃或拒绝。

令牌桶算法是一个存放固定容量令牌（token）的桶，按照固定速率往桶里添加令牌。令牌桶算法基本可以用下面的几个概念来描述：

• 令牌将按照固定的速率被放入令牌桶中。比如每秒放10个。
• 桶中最多存放b个令牌，当桶满时，新添加的令牌被丢弃或拒绝。
• 当一个n个字节大小的数据包到达，将从桶中删除n个令牌，接着数据包被发送到网络上。
• 如果桶中的令牌不足n个，则不会删除令牌，且该数据包将被限流（要么丢弃，要么缓冲区等待）。

令牌算法是根据放令牌的速率去控制输出的速率，也就是上图的to network的速率。to network我们可以理解为消息的处理程序，执行某段业务或者调用某个RPC。

​通俗的理解，令牌桶是一个水桶，而令牌是通过一根水管流到水桶中的水

    ​    ​令牌桶的填满时间，是由桶的自身容量、令牌漏出速率（桶下面的水管）、超过平均速率的突发流量持续的时间三个方面共同决定的。如果突发流量的时间比较短，令牌桶不会溢出，在通信流上不会受到影响，如果突发流量比较大，时间比较长，那令牌桶就会溢出，多余的通信流就会被限制。

 

​令牌桶算法和漏桶算法的区别

　　主要区别在于“漏桶算法”能够强行限制数据的传输速率，而“令牌桶算法”在能够限制数据的平均传输速率外，还允许某种程度的突发传输。在“令牌桶算法”中，只要令牌桶中存在令牌，那么就允许突发地传输数据直到达到用户配置的门限，因此它适合于具有突发特性的流量。

 

redis 实现

"local function addToQueue(x, time) "
+ " local count = 0 "
+ " for i = 1, x, 1 do "
+ " redis.call('lpush', KEYS[1], time) "
+ " count = count + 1 "
+ " end "
+ " return count "
+ "end "
+ "local result = 0 "
+ "local timeBase = redis.call('lindex', KEYS[1], tonumber(ARGV[2]) - tonumber(ARGV[1])) "
+ "if (timeBase == false) or (tonumber(ARGV[4]) - tonumber(timeBase) > tonumber(ARGV[3])) then "
+ " result = result + addToQueue(tonumber(ARGV[1]), tonumber(ARGV[4])) "
+ "end "
+ "if (timeBase ~= false) then "
+ " redis.call('ltrim', KEYS[1], 0, tonumber(ARGV[2])) "
+ "end "
+ "return result";

 

使用RateLimiter完成简单的大流量限流

    import com.google.common.util.concurrent.RateLimiter;  
      
    import java.util.ArrayList;  
    import java.util.List;  
    import java.util.concurrent.ExecutorService;  
    import java.util.concurrent.Executors;  
      
    /** 
     *
     * 有很多个任务，但希望每秒不超过X个，可用此类 
     */  
    public class Demo {  
      
        public static void main(String[] args) {  
            //0.5代表一秒最多多少个  
            RateLimiter rateLimiter = RateLimiter.create(0.5);  
            List<Runnable> tasks = new ArrayList<Runnable>();  
            for (int i = 0; i < 10; i++) {  
                tasks.add(new UserRequest(i));  
            }  
            ExecutorService threadPool = Executors.newCachedThreadPool();  
            for (Runnable runnable : tasks) {  
                System.out.println("等待时间：" + rateLimiter.acquire());  
                threadPool.execute(runnable);  
            }  
        }  
      
        private static class UserRequest implements Runnable {  
            private int id;  
      
            public UserRequest(int id) {  
                this.id = id;  
            }  
      
            public void run() {  
                System.out.println(id);  
            }  
        }  
      
    }