漏桶算法与令牌桶限流算法实现

1
2
3

X-RateLimit-Limit: 60         //每秒60次请求
X-RateLimit-Remaining: 23     //当前还剩下多少次
X-RateLimit-Reset: 1540650789 //限制重置时间

1

limiter := NewLimiter(10, 1);

1
2

limit := Every(100 * time.Millisecond);
limiter := NewLimiter(limit, 1);

1
2

func (lim *Limiter) Wait(ctx context.Context) (err error)
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)

1
2

func (lim *Limiter) Allow() bool
func (lim *Limiter) AllowN(now time.Time, n int) bool

1
2

func (lim *Limiter) Reserve() *Reservation
func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation

1
2
3
4
5
6
7

r := lim.Reserve()
f !r.OK() {
    // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
    return
}
time.Sleep(r.Delay())
Act() // 执行相关逻辑

1

const Inf = Limit(math.MaxFloat64)

1

const InfDuration = time.Duration(1<<63 - 1)

1

type Limit float64

1

func Every(interval time.Duration) Limit

1
2
3

type Limiter struct {
    // contains filtered or unexported fields
}

1

func NewLimiter(r Limit, b int) *Limiter

1

func (lim *Limiter) Allow() bool

1

func (lim *Limiter) AllowN(now time.Time, n int) bool

1

func (lim *Limiter) Burst() int

1

func (lim *Limiter) Limit() Limit

1

func (lim *Limiter) Reserve() *Reservation

1

func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation

1
2
3
4
5
6
7

r := lim.ReserveN(time.Now(), 1)
if !r.OK() {
  // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
  return
}
time.Sleep(r.Delay())
Act()

1

func (lim *Limiter) SetBurst(newBurst int)

1

func (lim *Limiter) SetBurstAt(now time.Time, newBurst int)

1

func (lim *Limiter) SetLimit(newLimit Limit)

1

func (lim *Limiter) SetLimitAt(now time.Time, newLimit Limit)

1

func (lim *Limiter) Wait(ctx context.Context) (err error)

1

func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)

1
2
3

type Reservation struct {
    // contains filtered or unexported fields
}

1

func (r *Reservation) Cancel()

1

func (r *Reservation) CancelAt(now time.Time)

1

func (r *Reservation) Delay() time.Duration

1

func (r *Reservation) DelayFrom(now time.Time) time.Duration

1

func (r *Reservation) OK() bool

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

package main

import (
	"context"
	"fmt"
	"golang.org/x/time/rate"
	"time"
)

func main() {
	l := rate.NewLimiter(1, 3) // 一个参数为每秒发生多少次事件，第二个参数是最大可运行多少个事件(burst)
	c, _ := context.WithCancel(context.TODO())
	for {
		l.Wait(c)
		fmt.Println(time.Now().Format("04:05.000"))
	}
}

1
2
3
4
5
6

25:08.243
25:08.243
25:08.243
25:09.248
25:10.248
25:11.248

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

package main

import (
	"fmt"
	"golang.org/x/time/rate"
	"time"
)

func main() {
	l := rate.NewLimiter(1, 3) // 一个参数为每秒发生多少次事件，第二个参数是最大可运行多少个事件(burst)
	for {
		if l.AllowN(time.Now(), 1) {
			fmt.Println(time.Now().Format("04:05.000"))
		} else {
			time.Sleep(1 * time.Second / 10)
			fmt.Println(time.Now().Format("Second 04:05.000"))
		}

	}
}

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

package main

import (
	"fmt"
	"golang.org/x/time/rate"
	"time"
)

func main() {
	l := rate.NewLimiter(1, 3) // 一个参数为每秒发生多少次事件，第二个参数是最大可运行多少个事件(burst)
	for {
		r := l.ReserveN(time.Now(), 1)
		s := r.Delay()
		time.Sleep(s)
		fmt.Println(s, time.Now().Format("04:05.000"))

	}
}

 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

package main

import (
	"bytes"
	"fmt"
	"io"
	"time"

	"golang.org/x/time/rate"
)

type reader struct {
	r       io.Reader
	limiter *rate.Limiter
}

// Reader returns a reader that is rate limited by
// the given token bucket. Each token in the bucket
// represents one byte.
func NewReader(r io.Reader, l *rate.Limiter) io.Reader {
	return &reader{
		r:       r,
		limiter: l,
	}
}

func (r *reader) Read(buf []byte) (int, error) {
	n, err := r.r.Read(buf)
	if n <= 0 {
		return n, err
	}
	now := time.Now()
	rv := r.limiter.ReserveN(now, n)
	if !rv.OK() {
		return 0, fmt.Errorf("Exceeds limiter's burst")
	}
	delay := rv.DelayFrom(now)
	fmt.Printf("Read %d bytes, delay %f\n", n, delay.Seconds())
	time.Sleep(delay)
	return n, err
}

func main() {
	// Source holding 1MB
	src := bytes.NewReader(make([]byte, 1024*1024))
	// Destination
	dst := &bytes.Buffer{}

	// Bucket adding 100KB every second, holding max 100KB
	limit := rate.NewLimiter(100*1024, 100*1024)

	start := time.Now()

	buf := make([]byte, 10*1024)
	// Copy source to destination, but wrap our reader with rate limited one
	//io.CopyBuffer(dst, NewReader(src, limit), buf)
	r := NewReader(src, limit)
	for {
		if n, err := r.Read(buf); err == nil {
			dst.Write(buf[0:n])
		} else {
			break
		}
	}
	fmt.Printf("Copied %d bytes in %s\n", dst.Len(), time.Since(start))
}

1
2
3
4
5
6

//定义某个时间的最大频率
//表示每秒的事件数
type Limit float64

//Inf表示无速率限制
const Inf = Limit(math.MaxFloat64)

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

type Limiter struct {
	limit Limit  //每秒允许处理的事件数量，即每秒处理事件的频率
	burst int   //令牌桶的最大数量， 如果burst为0，则除非limit == Inf，否则不允许处理任何事件。

	mu     sync.Mutex
	tokens float64  //令牌桶中可用的令牌数量
	// last is the last time the limiter's tokens field was updated
	//记录上次limiter的tokens被更新的时间
	last time.Time
	// lastEvent is the latest time of a rate-limited event (past or future)
	//lastEvent记录速率受限制(桶中没有令牌)的时间点，该时间点可能是过去的，也可能是将来的(Reservation预定的结束时间点)
	lastEvent time.Time
}

1
2
3
4
5
6
7
8

type Reservation struct {
	ok        bool //到截至时间是否可以获取足够的令牌
	lim       *Limiter
	tokens    int   //需要获取的令牌数量
	timeToAct time.Time  //需要等待的时间点
	// This is the Limit at reservation time, it can change later.
	limit Limit
}

1
2
3
4
5
6

func NewLimiter(r Limit, b int) *Limiter {
	return &Limiter{
		limit: r,
		burst: b,
	}
}

1
2
3
4
5
6

func Every(interval time.Duration) Limit {
	if interval <= 0 {
		return Inf
	}
	return 1 / Limit(interval.Seconds())
}

1
2
3

func (lim *Limiter) Allow() bool {
	return lim.AllowN(time.Now(), 1)
}

1
2
3

func (lim *Limiter) AllowN(now time.Time, n int) bool {
	return lim.reserveN(now, n, 0).ok
}

1
2
3

func (lim *Limiter) Wait(ctx context.Context) (err error) {
	return lim.WaitN(ctx, 1)
}

 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

func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
	//同步获取令牌桶的最大容量burst和限流器的速率limit
	lim.mu.Lock()
	burst := lim.burst
	limit := lim.limit
	lim.mu.Unlock()

	//如果n大于令牌桶的最大容量，则返回error
	if n > burst && limit != Inf {
		return fmt.Errorf("rate: Wait(n=%d) exceeds limiter's burst %d", n, lim.burst)
	}
	// Check if ctx is already cancelled
	//判断上下文ctx是否已经被取消，如果已经取消则返回error
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	// Determine wait limit
	now := time.Now()
	waitLimit := InfDuration
	if deadline, ok := ctx.Deadline(); ok { //如果可以获取上下文的截至时间，则更新可以等待的时间waitLimit
		waitLimit = deadline.Sub(now)
	}

	// Reserve
	//调用reserveN获取Reversation
	r := lim.reserveN(now, n, waitLimit)
	if !r.ok { //没有足够的时间获取令牌，则返回error
		return fmt.Errorf("rate: Wait(n=%d) would exceed context deadline", n)
	}
	// Wait if necessary
	//需要等待的时间
	delay := r.DelayFrom(now)
	if delay == 0 {
		return nil
	}
	t := time.NewTimer(delay)
	defer t.Stop()
	select {
	case <-t.C:
		// We can proceed.
		return nil
	case <-ctx.Done():
		// Context was canceled before we could proceed.  Cancel the
		// reservation, which may permit other events to proceed sooner.
		r.Cancel()
		return ctx.Err()
	}
}

1
2
3
4
5
6
7
8
9

// elapsed 表示过去的时间差
elapsed := now.Sub(lim.last)
// delta 表示这段时间一共新产生了多少 Token
delta = tokensFromDuration(now.Sub(lim.last))

tokens := lim.tokens + delta
if(token> lim.burst){
	token = lim.burst
}

 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

// now 是传入的当前的时间点，返回的 newNow 其实就是传入的参数，没有任何改变
// newLast 是更新 token 的时间
// newTokens 是 token 的数量
func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
	//last不能在当前时间now之后，否则计算出来的elapsed为负数，会导致令牌桶数量减少
	last := lim.last
	if now.Before(last) {
		last = now
	}

	// Avoid making delta overflow below when last is very old.
	//根据令牌桶的缺数计算出令牌桶未进行更新的最大时间
	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.
	//根据未更新的时间(未向桶中加入令牌的时间段)计算出产生的令牌数。
	delta := lim.limit.tokensFromDuration(elapsed)
	tokens := lim.tokens + delta  //计算出可用的令牌数
	if burst := float64(lim.burst); tokens > burst {
		tokens = burst
	}

	return now, last, tokens
}

 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

// now: 需要消费 token 的时间点
// n: 需要多少个 token
// maxFutureReserve: 能够等待的最长时间
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减去需要获取的令牌数(n)
	tokens -= float64(n)

	// Calculate the wait duration
	//如果tokens小于0，则说明桶中没有足够的令牌，计算出产生这些缺数的令牌需要多久(waitDuration)
	//计算出产生出缺数的令牌(即-tokens)需要多长时间
	var waitDuration time.Duration
	if tokens < 0 {
		waitDuration = lim.limit.durationFromTokens(-tokens)
	}

	// Decide result
	//如果n小于等于令牌桶的容量，并且可以等待到足够的令牌(即 waitDuration <= maxFutureReserve),则ok为true。表示可以获取到足够的令牌
	ok := n <= lim.burst && waitDuration <= maxFutureReserve

	// Prepare reservation
	r := Reservation{
		ok:    ok,
		lim:   lim,
		limit: lim.limit,
	}
	// 如果可以的话，就把 token 分配给预约者
	if ok {
		r.tokens = n // 需要的令牌数
		r.timeToAct = now.Add(waitDuration)  //计算获取到足够令牌的结束时间点
	}

	// Update state
	// 更新各个字段的状态
	if ok {
		lim.last = now  //更新tokens的时间
		lim.tokens = tokens  //更新令牌桶目前可用的令牌数tokens
		lim.lastEvent = r.timeToAct  //下次事件时间(即获取到足够令牌的时刻)
	} else {
		//last不能在当前时间now之后，否则计算出来的elapsed为负数，会导致令牌桶数量减少
		lim.last = last
	}

	lim.mu.Unlock()
	return r
}

1
2
3

func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	return d.Seconds() * float64(limit)
}

1
2
3
4
5
6
7
8
9

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

func (limit Limit) durationFromTokens(tokens float64) time.Duration {
	seconds := tokens / float64(limit)
	return time.Nanosecond *time.Duration(1e9*seconds)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

func (lim *Limiter) Reserve()*Reservation {
	return lim.ReserveN(time.Now(), 1)
}

// ReserveN returns a Reservation that indicates how long the caller must wait before n events happen.
// The Limiter takes this Reservation into account when allowing future events.
// ReserveN returns false if n exceeds the Limiter's burst size.
// Usage example:
//   r := lim.ReserveN(time.Now(), 1)
//   if !r.OK() {
//     // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
//     return
//   }
//   time.Sleep(r.Delay())
//   Act()
// Use this method if you wish to wait and slow down in accordance with the rate limit without dropping events.
// If you need to respect a deadline or cancel the delay, use Wait instead.
// To drop or skip events exceeding rate limit, use Allow instead.
func (lim *Limiter) ReserveN(now time.Time, n int)*Reservation {
	r := lim.reserveN(now, n, InfDuration)
	return &r
}

1
2
3

func (r *Reservation) OK() bool {
	return r.ok
}

1
2
3

func (r *Reservation) Delay() time.Duration {
	return r.DelayFrom(time.Now())
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

func (r *Reservation) DelayFrom(now time.Time) time.Duration {
	if !r.ok { //ok为false，则返回InfDuration
		return InfDuration
	}
	delay := r.timeToAct.Sub(now) //截止时间
	if delay < 0 { //如果截至时间已过，则返回0
		return 0
	}
	return delay
}

 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

func (r *Reservation) Cancel() {
	r.CancelAt(time.Now())
	return
}

func (r *Reservation) CancelAt(now time.Time) {
	if !r.ok {
		return
	}

	r.lim.mu.Lock()
	defer r.lim.mu.Unlock()

	/*
	1. 如果无需限流
	2. tokens为0 (需要获取的令牌数量为0)
	3. 已经过了截至时间
	以上三种情况无需处理取消操作
	*/
	if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(now) {
		return
	}

	// calculate tokens to restore
	// The duration between lim.lastEvent and r.timeToAct tells us how many tokens were reserved
	// after r was obtained. These tokens should not be restored.
	// 计算出需要还原的令牌数量
	// 这里的r.lim.lastEvent可能是本次Reservation的结束时间，也可能是后来的Reservation的结束时间，所以要把本次结束时间点(r.timeToAct)之后产生的令牌数减去
	restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))
	if restoreTokens <= 0 {
		return
	}


	// advance time to now
	//从新计算令牌桶的状态
	now, _, tokens := r.lim.advance(now)
	// calculate new number of tokens
	//还原当前令牌桶的令牌数量，当前的令牌数tokens加上需要还原的令牌数restoreTokens
	tokens += restoreTokens
	//如果tokens大于桶的最大容量，则将tokens置为桶的最大容量
	if burst := float64(r.lim.burst); tokens > burst {
		tokens = burst
	}
	// update state
	r.lim.last = now  //记录桶的更新时间
	r.lim.tokens = tokens //更新令牌数量
	//还原lastEvent，即上次速率受限制的时间
	if r.timeToAct == r.lim.lastEvent {
		prevEvent := r.timeToAct.Add(r.limit.durationFromTokens(float64(-r.tokens)))
		if !prevEvent.Before(now) {
			r.lim.lastEvent = prevEvent
		}
	}

	return
}

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

func reserveBug() {
	l := rate.NewLimiter(1, 1)
	for index := 0; index < 10; index++ {
		wg.Add(1)
		go func() {
			r := l.ReserveN(time.Now(), 1)
			time.Sleep(200 * time.Millisecond)
			r.Cancel()
			wg.Done()
		}()
	}

	wg.Wait()
	r := l.ReserveN(time.Now(), 1)
	fmt.Println("reserve need wait: ", r.Delay())
}

1
2
3
4
5
6
7
8

rl := ratelimit.New(100) // per second

prev := time.Now()
for i := 0; i < 10; i++ {
  now := rl.Take()
  fmt.Println(i, now.Sub(prev))
  prev = now
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// Output:
// 0 0
// 1 10ms
// 2 10ms
// 3 10ms
// 4 10ms
// 5 10ms
// 6 10ms
// 7 10ms
// 8 10ms
// 9 10ms

1
2
3
4
5
6
7
8

type Clock
type Limiter
    func New(rate int, opts ...Option) Limiter
    func NewUnlimited() Limiter
type Option
    func Per(per time.Duration) Option
    func WithClock(clock Clock) Option
    func WithSlack(slack int) Option

1
2
3
4

type Clock interface {
	Now() time.Time
	Sleep(time.Duration)
}

1
2
3
4

type Limiter interface {
	// Take should block to make sure that the RPS is met.
	Take() time.Time
}

1
2
3
4
5
6
7
8

type mutexLimiter struct {
	sync.Mutex
	last       time.Time
	sleepFor   time.Duration
	perRequest time.Duration
	maxSlack   time.Duration
	clock      Clock
}

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

type state struct {
	last     time.Time
	sleepFor time.Duration
}

type atomicLimiter struct {
	state unsafe.Pointer
	//lint:ignore U1000 Padding is unused but it is crucial to maintain performance
	// of this rate limiter in case of collocation with other frequently accessed memory.
	padding [56]byte // cache line size - state pointer size = 64 - 8; created to avoid false sharing.

	perRequest time.Duration
	maxSlack   time.Duration
	clock      Clock
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// New returns a Limiter that will limit to the given RPS.
func newMutexBased(rate int, opts ...Option) Limiter {
	l := &mutexLimiter{
		perRequest: time.Second / time.Duration(rate),
		maxSlack:   -10 * time.Second / time.Duration(rate),	// 最大松弛量
	}
	if l.clock == nil {
		l.clock = clock.New()
	}
	return l
}

1
2
3
4
5
6
7

sleepFor = t.perRequest - now.Sub(t.last)
if sleepFor > 0 {
	t.clock.Sleep(sleepFor)
	t.last = now.Add(sleepFor)
} else {
	t.last = now
}

 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

// Take blocks to ensure that the time spent between multiple
// Take calls is on average time.Second/rate.
func (t *mutexLimiter) Take() time.Time {
	t.Lock()
	defer t.Unlock()

	now := t.clock.Now()

	// If this is our first request, then we allow it.
	// 如果是第一次请求, 直接放行即可
	if t.last.IsZero() {
		t.last = now
		return t.last
	}

	// 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.
	// 注意这段代码，累加 sleepFor 的值
	t.sleepFor += t.perRequest - now.Sub(t.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.
	if t.sleepFor < t.maxSlack {
		t.sleepFor = t.maxSlack
	}

	// If sleepFor is positive, then we should sleep now.
	// 判断是否桶溢出. 如果桶溢出了, 需要 sleep 一段时间
	if t.sleepFor > 0 {
		t.clock.Sleep(t.sleepFor)
		t.last = now.Add(t.sleepFor)
		t.sleepFor = 0
	} else {
		t.last = now
	}

	return t.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
49
50
51
52
53
54
55
56

func (t *atomicLimiter) Take() time.Time {
	var (
        // 状态
		newState state
        // 用于表示原子操作是否成功
		taken    bool
        // 需要 sleep 的时间
		interval time.Duration
	)

    // 如果 CAS 操作不成功就一直尝试
	for !taken {
        // 获取当前的时间
		now := t.clock.Now()

        // load 出上一次调用的时间
		previousStatePointer := atomic.LoadPointer(&t.state)
		oldState := (*state)(previousStatePointer)

		newState = state{
			last:     now,
			sleepFor: oldState.sleepFor,
		}

		// 如果 last 是零值的话，表示之前就没用过，直接保存返回即可
		if oldState.last.IsZero() {
			taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
			continue
		}

		// sleepFor 是需要睡眠的时间，由于引入了松弛时间，所以 sleepFor 可能是一个
        // maxSlack ~ 0 之间的一个值，所以这里需要将现在的需要 sleep 的时间和上一次
        // sleepFor 的值相加
		newState.sleepFor += t.perRequest - now.Sub(oldState.last)

        // 如果距离上一次调用已经很久了，sleepFor 可能会是一个很小的值
        // 最小值只能是 maxSlack 的大小
		if newState.sleepFor < t.maxSlack {
			newState.sleepFor = t.maxSlack
		}

        // 如果 sleepFor 大于 0  的话，计算出需要 sleep 的时间
        // 然后将 state.sleepFor 置零
		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))
	}

    // sleep interval
	t.clock.Sleep(interval)
	return newState.last
}

1
2
3
4
5
6
7
8

t.sleepFor += t.perRequest - now.Sub(t.last)
if t.sleepFor > 0 {
  t.clock.Sleep(t.sleepFor)
  t.last = now.Add(t.sleepFor)
  t.sleepFor = 0
} else {
  t.last = now
}

1
2
3

if t.sleepFor < t.maxSlack {
  t.sleepFor = t.maxSlack
}

1

func New(rate int, opts ...Option) Limiter

1
2

limiter := ratelimit.New(100, ratelimit.WithoutSlack)
WithClock(clock Clock)

1
2
3
4
5
6

type Clock interface {
	Now() time.Time
	Sleep(time.Duration)
}
clock &= MyClock{}
limiter := ratelimit.New(100, ratelimit.WithClock(clock))