浅谈限流

 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

// Take blocks to ensure that the time spent between multiple
// Take calls is on average time.Second/rate.
func (t *atomicLimiter) Take() time.Time {
	var (
		newState state
		taken    bool
		interval time.Duration
	)
	for !taken {
		now := t.clock.Now()
		// 获取上一次请求的状态
		previousStatePointer := atomic.LoadPointer(&t.state)
		oldState := (*state)(previousStatePointer)

		// 根据上次请求的状态设置新的状态，这里的 sleepFor 可以理解为之前桶内还剩多少，sleepFor <= 0
		newState = state{
			last:     now,
			sleepFor: oldState.sleepFor,
		}

		// 首次请求直接放行
		if oldState.last.IsZero() {
			taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
			continue
		}
		// sleepFor calculates how much time we should sleep based on
		// the perRequest budget and how long the last request took.
		// Since the request may take longer than the budget, this number
		// can get negative, and is summed across requests.
		// 同时实现将之前剩余的减掉，和将这次剩余的加到桶中两种操作
		newState.sleepFor += t.perRequest - now.Sub(oldState.last)
		// We shouldn't allow sleepFor to get too negative, since it would mean that
		// a service that slowed down a lot for a short period of time would get
		// a much higher RPS following that.
		// maxSlack 可以理解为桶大小，可以应对一定突发流量。
		if newState.sleepFor < t.maxSlack {
			newState.sleepFor = t.maxSlack
		}
		// 有剩余可以直接放行
		if newState.sleepFor > 0 {
			newState.last = newState.last.Add(newState.sleepFor)
			interval, newState.sleepFor = newState.sleepFor, 0
		}
		taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
	}
	//阻塞等待
	t.clock.Sleep(interval)
	return newState.last
}

 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

// advance calculates and returns an updated state for lim resulting from the passage of time.
// lim is not changed.
// advance requires that lim.mu is held.
// 根据当前时间计算当前桶的状态
func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
	last := lim.last
	if now.Before(last) {
		last = now
	}

	// Avoid making delta overflow below when last is very old.
	// 先计算桶内还能容纳多少，避免两个请求间间隔太长的情况下，将间隔转化为 tokens 后溢出，采取用时间间隔来判断的方法来避免溢出。
	maxElapsed := lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)
	elapsed := now.Sub(last)
	if elapsed > maxElapsed {
		elapsed = maxElapsed
	}

	// Calculate the new number of tokens, due to time that passed.
	// 根据时间间隔计算 token, 更新桶内 token 数
	delta := lim.limit.tokensFromDuration(elapsed)
	tokens := lim.tokens + delta
	if burst := float64(lim.burst); tokens > burst {
		tokens = burst
	}

	return now, last, tokens
}

// durationFromTokens is a unit conversion function from the number of tokens to the duration
// of time it takes to accumulate them at a rate of limit tokens per second.
// 将 token 转换为时间
func (limit Limit) durationFromTokens(tokens float64) time.Duration {
	seconds := tokens / float64(limit)
	return time.Nanosecond * time.Duration(1e9*seconds)
}

// tokensFromDuration is a unit conversion function from a time duration to the number of tokens
// which could be accumulated during that duration at a rate of limit tokens per second.
// 将时间转换为 token
func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	// Split the integer and fractional parts ourself to minimize rounding errors.
	// See golang.org/issues/34861.
	// 分别对整数部分和小数部分计算，提高精度
	sec := float64(d/time.Second) * float64(limit)
	nsec := float64(d%time.Second) * float64(limit)
	return sec + nsec/1e9
}

 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

// reserveN is a helper method for AllowN, ReserveN, and WaitN.
// maxFutureReserve specifies the maximum reservation wait duration allowed.
// reserveN returns Reservation, not *Reservation, to avoid allocation in AllowN and WaitN.
func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation {
	lim.mu.Lock()

	// 未设置限流 直接通过
	if lim.limit == Inf {
		lim.mu.Unlock()
		return Reservation{
			ok:        true,
			lim:       lim,
			tokens:    n,
			timeToAct: now,
		}
	}

	// 计算当前桶内有多少 tokens
	now, last, tokens := lim.advance(now)

	// Calculate the remaining number of tokens resulting from the request.
	// 减去需要消耗的 tokens
	tokens -= float64(n)

	// Calculate the wait duration
	var waitDuration time.Duration
	// tokens 不够，计算要等待的时间
	if tokens < 0 {
		waitDuration = lim.limit.durationFromTokens(-tokens)
	}

	// Decide result
	// 请求量没有超过桶大小 并且等待时间小于等于最大等待时间则可以放行
	ok := n <= lim.burst && waitDuration <= maxFutureReserve

	// Prepare reservation
	// 更新 Reservation
	r := Reservation{
		ok:    ok,
		lim:   lim,
		limit: lim.limit,
	}
	if ok {
		r.tokens = n
		r.timeToAct = now.Add(waitDuration)
	}

	// Update state
	// 更新令牌桶状态
	if ok {
		lim.last = now
		lim.tokens = tokens
		lim.lastEvent = r.timeToAct
	} else {
		lim.last = last
	}

	lim.mu.Unlock()
	return r
}

1
2
3
4
5
6
7
8
9

type DynamicRateLimiter struct {
	limiter      *rate.Limiter
	curRate      int  //当前速率
	inflight     int  //当前在途请求数
	lastInfligt  int  //上个时间窗口的在途请求数
	requests     int  //当前时间窗口请求总数 用来判断是否是空闲情况
	fromOverload bool //是否已经经历了过载
	mu           sync.Mutex
}

1
2
3
4
5
6

func NewDynamicRateLimiter() *DynamicRateLimiter {
	// 初始速率为 100QPS 桶大小为 10 初始可以通过的最大 QPS 为 110QPS, 之后根据服务状态动态调整
	drl := &DynamicRateLimiter{curRate: 100, limiter: rate.NewLimiter(rate.Every(1/10.0*1000*time.Millisecond), 10)}
	go drl.watch()
	return drl
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

//Allow 方法在请求进来时调用，统计在途请求数，总请求数
func (d *DynamicRateLimiter) Allow() bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.inflight++
	d.requests++
	return d.limiter.Allow()
}

//Done 方法在请求完成后调用，将在途请求数减去
func (d *DynamicRateLimiter) Done() {
	d.mu.Lock()
	d.inflight--
	d.mu.Unlock()
}

 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

func (d *DynamicRateLimiter) watch() {
	for {
		//收集 CPU 使用率指标
		percent, _ := cpu.Percent(0, false)

		d.mu.Lock()
		// CPU 使用率小于 80%, 认为没有过载
		if percent[0] < 80 {
			// 根据请求数判断空闲状态，防止在空闲状态下请求速率一直增加
			// 在途请求数减少的情况说明服务还有空闲的能力可以进行服务
			// 即使服务状态一直好也不能一直增加，避免溢出
			if d.requests > 0 && (d.inflight == 0 || d.lastInfligt >= d.inflight) && d.curRate < 10000 {
				if !d.fromOverload { // 没有经历过过载 慢启动 指数增长
					d.curRate *= 2
				} else { // 经历过载以后 可能只是在阈值下面一点点，此时应缓慢增加，防止增加超出太多，引起服务崩溃
					d.curRate += int(0.2 * float64(d.curRate))
				}
			}
		} else {
			// CPU 使用率超过 80% 并且在途请求数增多，说明系统的处理能力已经达到了瓶颈，有很多阻塞的请求，此时缓慢减少，并设置过载标识
			if d.lastInfligt < d.inflight {
				d.curRate -= int(0.2 * float64(d.curRate))
				d.fromOverload = true
			}
		}

		// 监控统计
		limitGauge.Set(float64(d.curRate))
		cpuUsageGauge.Set(percent[0])

		// 更新状态
		d.lastInfligt = d.inflight
		d.limiter.SetLimit(rate.Every(time.Duration(1/float64(d.curRate)*1000) * time.Millisecond))
		d.requests = 0
		d.mu.Unlock()

		// 每秒更新一次
		time.Sleep(1 * time.Second)
	}
}

 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

// Handler
func (h *Handler) hello(c echo.Context) error {
	allow := h.limiter.Allow()
	defer h.limiter.Done()
	if !allow {
		requestCounter.WithLabelValues("429").Inc()
		return c.JSON(429, nil)
	}
	var o TestObject
	if err := c.Bind(&o); err != nil {
		return err
	}
	fib(int(randN(30, 35)))
	requestCounter.WithLabelValues("200").Inc()
	return c.JSON(200, o)
}

func main() {
	rand.Seed(time.Now().UnixNano())
	go func() {
		http.Handle("/metrics", promhttp.Handler())
		http.ListenAndServe(":6060", nil)
	}()
	// Echo instance
	e := echo.New()

	// Middleware
	e.Use(middleware.Logger())
	e.Use(middleware.Recover())

	h := NewHandler()
	// Routes
	e.POST("/", h.hello)

	// Start server
	e.Logger.Fatal(e.Start(":1323"))
}