常用限流算法

1.计数器法

设置一个计数器counter，在一段时间内（假定60秒）每当一个请求过来的时候counter加1，如果counter超过设定的阈值（假定100）就限流，过完60秒重置计数器。

该种方法实现起来很简单，但其实是有临界问题的，假如在第一分的后1s来了100个请求，第二分种的前1s来了100个请求，那在这1秒内（1:00）其实最大QPS为200。如下图：

计数器法会有临界问题，主要还是统计的精度太低，这点可以通过滑动窗口算法解决

2.滑动窗口(Sliding Window)

Sliding Window Algorithm(滑动窗口算法）是计数器算法的升级版，它的核心思想是把一段时间窗口划分成更小的时间块且整个时间窗口往前动态推进，每个时间块单独计数，最后统计这段动态时间窗口的总计数，以解决计数器算法中临界值的问题。

如图所示，整个红色的矩形框表示一个时间窗口，长度为一分钟。然后我们将时间窗口进行划分，比如图中我们就将滑动窗口划成了6格，所以每格代表的是10秒钟。每过10秒钟时间窗口就会往右滑动一格。每一个格都有自己独立的计数器counter，比如当一个请求在0:35秒的时候到达，那么0:30~0:39对应格的counter就会加1。

滑动窗口怎么解决刚才的临界问题的呢？

在上图中，0:59到达的100个请求会落在灰色的格子中，而1:00到达的请求会落在橘黄色的格子中。当时间到达1:00时，我们的窗口会往右移动一格，那么此时时间窗口内的总请求数量一共是200个，超过了限定的100个，所以此时能够检测出来触发了限流。

代码实现：

  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
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142


package main

import (
	"container/list"
	"fmt"
	"sync"
	"time"
)

//var Total int64

type Queue struct {
	sync.RWMutex
	data *list.List
}

func NewQueue() *Queue {
	q := new(Queue)
	q.data = list.New()
	return q
}

func (q *Queue) push(v interface{}) {
	defer q.Unlock()
	q.Lock()
	q.data.PushFront(v)
}

func (q *Queue) pop() interface{} {
	defer q.Unlock()
	q.Lock()
	iter := q.data.Back()
	v := iter.Value
	q.data.Remove(iter)
	return v
}

func (q *Queue) dump() {
	for iter := q.data.Back(); iter != nil; iter = iter.Prev() {
		fmt.Println("item:", iter.Value)
	}
}


type Count struct {
	sync.Mutex
	sum int64
}

type LastCount struct {
	sync.Mutex
	sum map[string]int64
}

type SlidingWindow struct {
	queue *Queue
	count Count
	lastCount LastCount
	//阈值
	limit int64
	//间隔
	period int
	//当前点
	cur string
}

func(s *SlidingWindow) incrCount(){
	s.count.Lock()
	defer s.count.Unlock()
	s.count.sum++
}

func(s *SlidingWindow) decrCount(v int64){
	s.count.Lock()
	defer s.count.Unlock()
	s.count.sum= s.count.sum - v
}

func(s *SlidingWindow) incrLastCount(k string){
	s.lastCount.Lock()
	defer s.lastCount.Unlock()
	s.lastCount.sum[k]++
}

func(s *SlidingWindow) restLastCount(k string){
	s.lastCount.Lock()
	defer s.lastCount.Unlock()
	s.lastCount.sum[k]=0
}


func main()  {
	sw:= &SlidingWindow{
		queue:     NewQueue(),
		count:     Count{},
		lastCount: LastCount{
			sum: make(map[string]int64),
		},
		limit:     80,
		period:    10,
		cur: "",
	}

	for i:=1;i<=sw.period;i++ {
		v:= fmt.Sprintf("queue_%d",i)
		sw.queue.push(v)
	}
	sw.cur="queue_10"

	ticker := time.NewTicker(1 * time.Second)
	go func(t *time.Ticker) {
		for {
			<-t.C
			last:=sw.queue.pop().(string)
			fmt.Println(last)
			c := sw.lastCount.sum[last]
			if  c > 0 {
				fmt.Println("窗口往前移动，减去窗口前的记数")
				sw.decrCount(c)
				sw.restLastCount(last)

			}

			sw.cur = last
			sw.queue.push(last)
		}
	}(ticker)

	for {
		fmt.Println(sw.count.sum)
		if sw.count.sum > sw.limit {
			fmt.Println("超限。。。。")
			time.Sleep(time.Second*3)
		}
		fmt.Println("正常访问。。。。")
		sw.incrCount()
		sw.incrLastCount(sw.cur)
		fmt.Println(sw.lastCount)
		time.Sleep(time.Millisecond*100)

	}
}

我再来回顾一下刚才的计数器算法，我们可以发现，计数器算法其实就是滑动窗口算法，只是它没有对时间窗口做进一步地划分，所以只有1格。由此可见，当滑动窗口的格子划分的越多，那么滑动窗口的移动就越平滑，限流的统计就会越精确，但相应的资源消耗也更高。

3.漏桶算法(Leaky Bucket)

漏桶算法思路是，有一个固定大小的桶，水（请求）忽快忽慢的进入到漏桶里，漏桶以一定的速度出水。当桶满了之后会发生溢出。

维基百科在对该算法定义了两种实现，分别是as a meter和as a queue，但是网上大多数文章都没有提到这一点。

As a meter

第一种实现是和令牌桶等价的，只是表述角度不同。

伪代码如下：

 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


package main

import (
	"fmt"
	"time"
)

func Max(x, y int64) int64 {
	if x > y {
		return x
	}
	return y
}

func currentTimeMillis() int64 {
	return time.Now().UnixNano() / int64(time.Millisecond)
}

func main() {
	//桶大小
	var bucketSize int64 = 10
	//每秒流出请求数
	var rate int64 =2
	//当前桶的水量
	var count int64 =0
	timeStamp := currentTimeMillis()

	for {
		now := currentTimeMillis()
		if(now > timeStamp) {
			count = Max(0,count-(now-timeStamp)*rate)
			timeStamp = now
		}

		if (count+1 <= bucketSize) {
			fmt.Println("正常")
			count++
		}else{
			fmt.Println("桶满了,暂停1秒")
			time.Sleep(time.Second)
		}
	}
}

该种实现允许一段时间内的突发流量，比如初始时桶中没有水，这时1ms内来了10个请求，这10个请求是不会被限流的，但之后每ms最多只能接受2个请求（比如下1ms又来了10个请求，那其中8个请求是会被限流的）。

其达到的效果和令牌桶一样。

As a queue

在这种实现中桶被当作FIFO缓冲区或队列用，具体可以看下维基百科中的解释。我的理解这种多应用于带宽、下载这类流量限速中，因为实际中数据包和产生速度都不是固定的，有大有小，有快有慢，经过桶的缓冲后以一个固定的速度往外传速，起到流量整形的作用。

4.令牌桶算法(Token Bucket)

令牌桶算法的思想就是，桶中最多有N个令牌，会以一定速率往桶中加令牌，每个请求都需要从令牌桶中取出相应的令牌才能放行，如果桶中没有令牌则被限流。

令牌桶算法与上文的漏桶算法as a meter实现是等价的，能够在限制数据的平均传输速率的同时还允许某种程度的突发传输。伪代码：

 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


package main

import (
	"fmt"
	"time"
)

func Min(x, y int64) int64 {
	if x < y {
		return x
	}
	return y
}

func main() {
	var token int64 =0
	var bucketSize int64 = 10
	var rate int64 =2
	timeStamp := time.Now().UnixNano() / int64(time.Millisecond)

	for {
		now:=time.Now().UnixNano() / int64(time.Millisecond)
		if(now > timeStamp) {
			fmt.Println("发token")
			token = Min(bucketSize, token+(now-timeStamp)*rate)
			timeStamp = now
		}

		if (token > 0) {
			fmt.Println("正常")
			token--
		}else{
			fmt.Println("限流了,暂停1秒")
			time.Sleep(time.Second)
		}
	}
}

5.漏桶算法两种实现和令牌桶算法的对比

as a meter的漏桶算法和令牌桶算法是一样的，只是思想角度有所不同。

as a queue的漏桶算法能强行限制数据的传输速率，而令牌桶和as a meter漏桶则能够在限制数据的平均传输速率的同时还允许某种程度的突发传输。

6.总结

从以下几个维度对各种限流算法做下对比：

精确性：满足“长度为 T 的时间内访问次数不超过 N”的程度
流量整形：减少输出流量的突刺、抖动，输出平整流量
突发流量：支持瞬时高访问量
算法实现复杂度：算法的复杂度
使用场景：①未达到系统最大承载能力，但出于保护的目的需要人为限流，如防爬限制某个接口的QPS,对外承诺某一接口的最高调用量。②当前系统的最大承载能力就这么多，在达到最大承载能力之前允许你有一定的突发，一旦达到就要限流以保护系统。

	精确性	输出抖动	突发流量	复杂度	使用场景
计数器	差	频繁	支持	简单	①
滑动窗口	一般（取决窗口粒度）	频繁	支持	简单	①
漏桶	高（取决桶的容量）	无	不支持	一般	①
令牌桶	高（取决桶的容量）	除突发流量外无抖动	支持	复杂	②