滑动窗口计算器如何对多个接口限流_分布式高并发系统限流原理与实践

// CounterLimit 计数器
type CounterLimit struct {
	counter      int64 //计数器
	limit        int64 //指定时间窗口内允许的最大请求数
	intervalNano int64 //指定的时间窗口
	unixNano     int64 //unix时间戳,单位为纳秒
}
 
// NewCounterLimit 初始化
func NewCounterLimit(interval time.Duration, limit int64) *CounterLimit {
 
	return &CounterLimit{
		counter:      0,
		limit:        limit,
		intervalNano: int64(interval),
		unixNano:     time.Now().UnixNano(),
	}
}
 
// Allow 判断当前时间窗口是否允许请求
func (c *CounterLimit) Allow() bool {
 
	now := time.Now().UnixNano()
	if now-c.unixNano > c.intervalNano { //如果当前过了当前的时间窗口,则重新进行计数
		atomic.StoreInt64(&c.counter, 0)
		atomic.StoreInt64(&c.unixNano, now)
		return true
	}
 
	atomic.AddInt64(&c.counter, 1)
	return c.counter < c.limit //判断是否要进行限流
}

var (
    limitCount  int        = 10 // 6s限频
    limitBucket int        = 6  // 滑动窗口个数
    curCount    int32      = 0  // 记录限频数量
    head        *ring.Ring      // 环形队列（链表）
)
 
func main() {
    tcpAddr, err := net.ResolveTCPAddr("tcp4", "0.0.0.0:9090") //获取一个tcpAddr
    checkError(err)
    listener, err := net.ListenTCP("tcp", tcpAddr) //监听一个端口
    checkError(err)
    defer listener.Close()
    // 初始化滑动窗口
    head = ring.New(limitBucket)
    for i := 0; i < limitBucket; i++ {
        head.Value = 0
        head = head.Next()
    }
    // 启动执行器
    go func() {
        timer := time.NewTicker(time.Second * 1)
        for range timer.C { // 定时每隔1秒刷新一次滑动窗口数据
            subCount := int32(0 - head.Value.(int))
            newCount := atomic.AddInt32(&curCount, subCount)
 
            arr := [6]int{}
            for i := 0; i < limitBucket; i++ { // 这里是为了方便打印
                arr[i] = head.Value.(int)
                head = head.Next()
            }
            fmt.Println("move subCount,newCount,arr", subCount, newCount, arr)
            head.Value = 0
            head = head.Next()
        }
    }()
 
    for {
        conn, err := listener.Accept() // 在此处阻塞，每次来一个请求才往下运行handle函数
        if err != nil {
            fmt.Println(err)
            continue
        }
        go handle(&conn) // 起一个单独的协程处理，有多少个请求，就起多少个协程，协程之间共享同一个全局变量limiting，对其进行原子操作。
    }
}
 
func handle(conn *net.Conn) {
    defer (*conn).Close()
    n := atomic.AddInt32(&curCount, 1)
    //fmt.Println("handler n:", n)
    if n > int32(limitCount) { // 超出限频
        atomic.AddInt32(&curCount, -1) // add 1 by atomic，业务处理完毕，放回令牌
        (*conn).Write([]byte("HTTP/1.1 404 NOT FOUNDrnrnError, too many request, please try again."))
    } else {
        mu := sync.Mutex{}
        mu.Lock()
        pos := head.Prev()
        val := pos.Value.(int)
        val++
        pos.Value = val
        mu.Unlock()
        time.Sleep(1 * time.Second)                                             // 假设我们的应用处理业务用了1s的时间
        (*conn).Write([]byte("HTTP/1.1 200 OKrnrnI can change the world!")) // 业务处理结束后，回复200成功。
    }
}
 
// 异常报错的处理
func checkError(err error) {
    if err != nil {
        fmt.Fprintf(os.Stderr, "Fatal error: %s", err.Error())
        os.Exit(1)
    }
}

$ ./hey -c 6 -n 300 -q 6 -t 80 http://localhost:9090
 
move subCount,newCount,arr 0 0 [0 0 0 0 0 0]
move subCount,newCount,arr 0 0 [0 0 0 0 0 0]
move subCount,newCount,arr 0 6 [0 0 0 0 0 6]
move subCount,newCount,arr 0 10 [0 0 0 0 6 4]
move subCount,newCount,arr 0 10 [0 0 0 6 4 0]
move subCount,newCount,arr 0 10 [0 0 6 4 0 0]
move subCount,newCount,arr 0 10 [0 6 4 0 0 0]
move subCount,newCount,arr -6 4 [6 4 0 0 0 0]
move subCount,newCount,arr -4 6 [4 0 0 0 0 6]
move subCount,newCount,arr 0 10 [0 0 0 0 6 4]
move subCount,newCount,arr 0 10 [0 0 0 6 4 0]
move subCount,newCount,arr 0 10 [0 0 6 4 0 0]
move subCount,newCount,arr 0 10 [0 6 4 0 0 0]
move subCount,newCount,arr -6 4 [6 4 0 0 0 0]
move subCount,newCount,arr -4 3 [4 0 0 0 0 3]
move subCount,newCount,arr 0 3 [0 0 0 0 3 0]
move subCount,newCount,arr 0 3 [0 0 0 3 0 0]
move subCount,newCount,arr 0 3 [0 0 3 0 0 0]
move subCount,newCount,arr 0 3 [0 3 0 0 0 0]
move subCount,newCount,arr -3 0 [3 0 0 0 0 0]
move subCount,newCount,arr 0 0 [0 0 0 0 0 0]
move subCount,newCount,arr 0 0 [0 0 0 0 0 0]

// BucketLimit 漏桶结构
type BucketLimit struct {
    rate       float64 //漏桶中水的漏出速率
    bucketSize float64 //漏桶最多能装的水大小
    unixNano   int64   //unix时间戳
    curWater   float64 //当前桶里面的水
}
 
// NewBucketLimit  初始化
func NewBucketLimit(rate float64, bucketSize int64) *BucketLimit {
    return &BucketLimit{
        bucketSize: float64(bucketSize),
        rate:       rate,
        unixNano:   time.Now().UnixNano(),
        curWater:   0,
    }
}
 
// refresh 更新当前桶的容量
func (b *BucketLimit) refresh() {
    now := time.Now().UnixNano()
    //时间差, 把纳秒换成秒
    diffSec := float64(now-b.unixNano) / 1000 / 1000 / 1000
    b.curWater = math.Max(0, b.curWater-diffSec*b.rate)
    b.unixNano = now
    return
}
 
// Allow 允许请求，是否超过桶的容量
func (b *BucketLimit) Allow() bool {
    b.refresh()
    if b.curWater < b.bucketSize {
        b.curWater = b.curWater + 1
        return true
    }
 
    return false
}

// TokenBucket 令牌桶
type TokenBucket struct {
    rate      int // 令牌放入速度
    tokenSize int // 令牌桶的容量
    curNum    int // 当前桶中token
}
 
// NewTokenBucket 初始化
func NewTokenBucket(rate, tokenSize int) *TokenBucket {
    return &TokenBucket{
        rate:      rate,
        tokenSize: tokenSize,
    }
}
 
// PushToken 在桶中存放token
func (t *TokenBucket) PushToken() chan struct{} {
    tb := make(chan struct{}, t.tokenSize)
    ticker := time.NewTicker(time.Duration(1000) * time.Microsecond)
    //初始化token
    for i := 0; i < t.tokenSize; i++ {
        tb <- struct{}{}
    }
    t.curNum = t.tokenSize
 
    // 指定速率放入token
    go func() {
        for {
            for i := 0; i < t.rate; i++ {
                tb <- struct{}{}
            }
            t.curNum += t.rate
            if t.curNum > t.tokenSize {
                t.curNum = t.tokenSize
            }
            <-ticker.C
        }
    }()
    return tb
}
 
// popToken 取出token
func (t *TokenBucket) PopToken(bucket chan struct{}, n int) {
    for i := 0; i < n; i++ {
        _, ok := <-bucket
        if ok {
            t.curNum -= 1
            fmt.Println("get  token  success")
        } else {
            fmt.Println("get  token  fail")
        }
    }
}