golang 多种限流器实现

2023-05-13 分类：GO 作者：admin 阅读（20）

限流

日常开发中，一般会遇到几种场景需要限流，比如有个api-server，需要限制单个用户的调用频率，避免用户恶意刷接口或者突发大流量导致服务不可用等，这边记录几个常用的限流方法。

并发控制

简单的并发控制，用户的所有请求丢到一个channel里，再指定一个带缓冲区的channel为并发池，缓冲池的长度就是可以同时存在的协程数量，然后将执行完的任务根据需要直接返回或者丢到另外一个channel 里, 这样做的问题是如果任务太多，后面的任务会慢慢等待(因为channel阻塞机制)，用户体验不是太好。

Code:

package main

import (
	"fmt"
	"sync"
)


func setJob() <-chan int {
    var wg sync.WaitGroup
    jobChan := make(chan int, 50)
    wg.Add(1)
    go func(){
        for i := 0; i < 50; i++ {
            jobChan <- i
        }
        close(jobChan)
        wg.Done()
    }()
    wg.Wait()
    return jobChan
}

func main() {
	var wg sync.WaitGroup
    // 创建一个需要处理的数据来源, 假设是个channel
    jobChan := setJob()

    // 将结果存入一个channel
    resChan := make(chan int , 50)

    // 设置一个并发池
	buckets := make(chan bool, 10)
	for job := range jobChan {
		buckets <- true
		wg.Add(1)
		go func(job int) {
            res := 10 * job
            resChan <- res
			<-buckets
			wg.Done()
		}(job)
	}

    wg.Wait()
    close(resChan)
    tmpA := []int{}
	for r := range(resChan) {
		tmpA = append(tmpA, r)
	}
    fmt.Println(tmpA)
    
}

package main

import (

"fmt"

"sync"

)

func setJob() <-chan int {

var wg sync.WaitGroup

jobChan := make(chan int, 50)

wg.Add(1)

go func(){

for i := 0; i < 50; i++ {

jobChan <- i

}

close(jobChan)

wg.Done()

}()

wg.Wait()

return jobChan

}

func main() {

var wg sync.WaitGroup

// 创建一个需要处理的数据来源, 假设是个channel

jobChan := setJob()

// 将结果存入一个channel

resChan := make(chan int , 50)

// 设置一个并发池

buckets := make(chan bool, 10)

for job := range jobChan {

buckets <- true

wg.Add(1)

go func(job int) {

res := 10 * job

resChan <- res

<-buckets

wg.Done()

}(job)

}

wg.Wait()

close(resChan)

tmpA := []int{}

for r := range(resChan) {

tmpA = append(tmpA, r)

}

fmt.Println(tmpA)

}

PS：先把所有请求都放入channel，至关重要。

漏桶

简单的限流，指定一个大小固定的桶(bucket),以指定速率流入流出，如果桶满了，则拒绝后续请求，好处是简单，但是由于rate是固定的，导致了在无法在业务突发高峰时候(比如活动期间)有比较好的适配性。

Code:

package main

import (
    "fmt"
    "math"
    "time"
)

// 漏桶限流器
type BucketLimit struct {
    rate       float64 //漏桶中水的漏出速率, 即每秒流多少水
    bucketSize float64 //漏桶最多能装的水大小
    lastAccessTime   time.Time //上次访问时间
    curWater   float64 //当前桶里面的水
}

func NewBucketLimit(rate float64, bucketSize int64) *BucketLimit {
    return &BucketLimit{
        bucketSize: float64(bucketSize),
        rate:       rate,
        curWater:   0,
    }
}

func (b *BucketLimit) AllowControl() bool {
    now := time.Now()
    pastTime := now.Sub(b.lastAccessTime)

    // 当前剩余水量，当前水量减去距离上次访问的流出水量，如果流完了，即剩余水量为0
    b.curWater = math.Max(0, b.curWater - float64(pastTime) * b.rate)

    b.lastAccessTime = now

    // 当前水量必须小于桶的总量，不然则流出了
    if b.curWater < b.bucketSize {
        b.curWater = b.curWater + 1
        return true
    }
    return false
}

func main() {
    // 创建一个流出速率为1qps,桶的总量为2的限流器
    limit := NewBucketLimit(1, 2)

    // 在桶里放入1000滴水
    for i := 0; i < 1000; i++ {
        allow := limit.AllowControl()
        if allow {
            fmt.Printf("第%d滴水, 顺利流出\n", i)
            continue
        } else {
            fmt.Printf("第%d滴水, 溢出丢弃\n", i)
            time.Sleep(time.Millisecond * 100)
        }
    }
}

package main

import (

"fmt"

"math"

"time"

)

// 漏桶限流器

type BucketLimit struct {

rate float64 //漏桶中水的漏出速率, 即每秒流多少水

bucketSize float64 //漏桶最多能装的水大小

lastAccessTime time.Time //上次访问时间

curWater float64 //当前桶里面的水

}

func NewBucketLimit(rate float64, bucketSize int64) *BucketLimit {

return &BucketLimit{

bucketSize: float64(bucketSize),

rate: rate,

curWater: 0,

}

func (b *BucketLimit) AllowControl() bool {

now := time.Now()

pastTime := now.Sub(b.lastAccessTime)

// 当前剩余水量，当前水量减去距离上次访问的流出水量，如果流完了，即剩余水量为0

b.curWater = math.Max(0, b.curWater - float64(pastTime) * b.rate)

b.lastAccessTime = now

// 当前水量必须小于桶的总量，不然则流出了

if b.curWater < b.bucketSize {

b.curWater = b.curWater + 1

return true

}

return false

}

func main() {

// 创建一个流出速率为1qps,桶的总量为2的限流器

limit := NewBucketLimit(1, 2)

// 在桶里放入1000滴水

for i := 0; i < 1000; i++ {

allow := limit.AllowControl()

if allow {

fmt.Printf("第%d滴水, 顺利流出\n", i)

continue

} else {

fmt.Printf("第%d滴水, 溢出丢弃\n", i)

time.Sleep(time.Millisecond * 100)

}

令牌桶

令牌桶相比漏桶可以对突发的流量做一定程度的处理，该方法意思是一定速率往一个桶里放令牌(token)，同时往桶里注水，每消耗一滴水，需要一定数量的令牌，这里代码假设需要2块令牌才能消耗1滴水，所以处理水滴的速度主要取决于令牌的数量。假设放入令牌的速度是4块/s，每秒消耗的水滴数量是1滴，即每秒消耗2块令牌，那么每秒多出来的2块令牌就会积累在桶里，直到桶满位置，然后因为某次活动大放水，每秒消耗的水滴数量突然变成了3滴，即每秒需要消耗6块另外，大于每秒放入的令牌数量，由于之前桶里有积累的令牌，所以即使放水量加大，依然可以在令牌桶消耗完之前快速处理。

Code:

package main
 
import (
    "fmt"
    "time"
    "math"
)
 
// 令牌桶限流器
type BucketLimit struct {
    rate       float64 //令牌桶放令牌的速率
    bucketSize float64 //令牌桶最多能装的令牌数量
    lastAccessTime   time.Time //上次访问时间
    curTokenNum  float64 //桶里当前的令牌数量
}

func NewBucketLimit(rate float64, bucketSize int64) *BucketLimit {
    return &BucketLimit{
        bucketSize: float64(bucketSize),
        rate:       rate,
        curTokenNum:   0,
    }
}

func (b *BucketLimit) AllowControl(tokenNeed float64) bool {
    now := time.Now()
    pastTime := now.Sub(b.lastAccessTime)
 
    // 在距离上次访问期间一共可发放了多少令牌
    newTokenNum := float64(pastTime) / float64(b.rate)

    // 剩余令牌数量不能超过桶的总空间
    b.curTokenNum = math.Min(b.bucketSize, b.curTokenNum + newTokenNum)

    b.lastAccessTime = now
    // tokenNeed 指处理一个请求需要的令牌数量
    if tokenNeed > b.curTokenNum {
        return false
    } else {
        b.curTokenNum = b.curTokenNum - tokenNeed
        return true
    }
}
 
func main() {
    // 创建一个放令牌速率为10qps,桶的总量为20的限流器
    limit := NewBucketLimit(10, 20)

    // 在桶里放入100滴水
    for i := 0; i < 100; i++ {
        // 处理1滴水需要消耗19块令牌
        if i == 50 {
            time.Sleep(3 * time.Second)
        }
        allow := limit.AllowControl(19)
        if allow {
            fmt.Printf("第%d滴水, 顺利流出\n", i)
            continue
        } else {
            fmt.Printf("第%d滴水, 溢出丢弃\n", i)
            time.Sleep(time.Millisecond * 100)
        }
    }
}

package main

import (

"fmt"

"time"

"math"

)

// 令牌桶限流器

type BucketLimit struct {

rate float64 //令牌桶放令牌的速率

bucketSize float64 //令牌桶最多能装的令牌数量

lastAccessTime time.Time //上次访问时间

curTokenNum float64 //桶里当前的令牌数量

}

func NewBucketLimit(rate float64, bucketSize int64) *BucketLimit {

return &BucketLimit{

bucketSize: float64(bucketSize),

rate: rate,

curTokenNum: 0,

}

func (b *BucketLimit) AllowControl(tokenNeed float64) bool {

now := time.Now()

pastTime := now.Sub(b.lastAccessTime)

// 在距离上次访问期间一共可发放了多少令牌

newTokenNum := float64(pastTime) / float64(b.rate)

// 剩余令牌数量不能超过桶的总空间

b.curTokenNum = math.Min(b.bucketSize, b.curTokenNum + newTokenNum)

b.lastAccessTime = now

// tokenNeed 指处理一个请求需要的令牌数量

if tokenNeed > b.curTokenNum {

return false

} else {

b.curTokenNum = b.curTokenNum - tokenNeed

return true

}

func main() {

// 创建一个放令牌速率为10qps,桶的总量为20的限流器

limit := NewBucketLimit(10, 20)

// 在桶里放入100滴水

for i := 0; i < 100; i++ {

// 处理1滴水需要消耗19块令牌

if i == 50 {

time.Sleep(3 * time.Second)

}

allow := limit.AllowControl(19)

if allow {

fmt.Printf("第%d滴水, 顺利流出\n", i)

continue

} else {

fmt.Printf("第%d滴水, 溢出丢弃\n", i)

time.Sleep(time.Millisecond * 100)

}

转自：https://blog.51cto.com/u_2010293/2781897

使用golang标准库实现令牌桶，更加简单：http://biji.jinli.vip/791.html

漏桶算法

请求处理：当漏桶满了（即存储请求的桶达到最大容量），新的请求会被丢弃或拒绝。这意味着在高流量时，可能会导致请求的丢失。
输出速率：请求以固定速率从桶中“漏出”，无论输入流量如何，输出都是恒定的。

令牌桶算法

请求处理：当桶中没有令牌时，请求不会被丢弃，而是会被阻塞或等待，直到有令牌可用。这意味着即使在高流量情况下，系统也不会丢失请求。
输出速率：请求的处理速率可以根据令牌的生成速率变化，允许短时间内处理更多请求，只要桶中有足够的令牌。

总结

漏桶算法：一旦桶满，新的请求会被丢弃，适用于需要严格控制流量的场景。
令牌桶算法：请求会被阻塞，直到有令牌可用，适用于需要处理突发流量的场景。

这种设计上的差异使得令牌桶算法在处理流量波动时更具灵活性，而漏桶算法则更适合于对流量有严格要求的情况。

「三年博客，如果觉得我的文章对您有用，请帮助本站成长」

赞(0) 打赏

标签：

上一篇：pod常用控制器介绍（deployment、StatefulSet、Job、CronJob、DaemonSet）

下一篇：使用k8s部署单节点etcd

你可能感兴趣

golang for range 遍历

golang 参考题

使用 pprof 排查 Go 死锁的步骤

Golang工程师技术题

golang 简易打印请求服务端

go defer和recover

go 协程池

go 中的json.Marshal

golang 常用工具方法

golang 实现简易协程池

共有 0 - golang 多种限流器实现

点击这里取消回复。

博客简介

admin（6年前 (2020-03-09)）
分别用不同厚度的筏板定义，画图后这设置筏板变截面处理。 http://f.fwxgx.co...
评：新文章！
admin（6年前 (2020-03-09)）
分别用不同厚度的筏板定义，画图后这设置筏板变截面处理。 http://f.fwxgx.co...
评：新文章！
admin（6年前 (2020-03-09)）
新增一个框架图！ http://biji.jinli.vip/wp-content/upl...
评：新文章！
一位WordPress评论者（6年前 (2020-02-13)）
嗨，这是一条评论。要开始审核、编辑及删除评论，请访问仪表盘的“评论”页面。评论者头像来自...
评：世界，您好！

golang 多种限流器实现

漏桶算法

令牌桶算法

总结

你可能感兴趣

共有 0 - golang 多种限流器实现

点击这里取消回复。

博客简介

最近更新

随机文章

热门标签

精彩评论