日常開發中，對于某些接口有請求頻率的限制。比如登錄的接口、發送短信的接口、秒殺商品的接口等等。

官方的golang.org/x/time/rate包中實現了令牌桶的算法。

封裝限流器可以將ip、手機號這種的作為限流器組的標識。

接下來就是實例化限流器和獲取令牌函數的實現

package limiter

//component/limiter/limiter.go

import (

? ? "sync"

? ? "time"

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

)

type Limiters struct {

? ? limiters map[string]*Limiter

? ? lock ? ? sync.Mutex

}

type Limiter struct {

? ? limiter *rate.Limiter

? ? lastGet time.Time //上一次獲取token的時間

? ? key ? ? string

}

var GlobalLimiters = &Limiters{

? ? limiters: make(map[string]*Limiter),

? ? lock: ? ? sync.Mutex{},

}

var once = sync.Once{}

func NewLimiter(r rate.Limit, b int, key string) *Limiter {

? ? once.Do(func() {

? ? ? ? go GlobalLimiters.clearLimiter()

? ? })

? ? keyLimiter := GlobalLimiters.getLimiter(r, b, key)

? ? return keyLimiter

}

func (l *Limiter) Allow() bool {

? ? l.lastGet = time.Now()

? ? return l.limiter.Allow()

}

func (ls *Limiters) getLimiter(r rate.Limit, b int, key string) *Limiter {

? ? ls.lock.Lock()

? ? defer ls.lock.Unlock()

? ? limiter, ok := ls.limiters[key]

? ? if ok {

? ? ? ? return limiter

? ? }

? ? l := &Limiter{

? ? ? ? limiter: rate.NewLimiter(r, b),

? ? ? ? lastGet: time.Now(),

? ? ? ? key: ? ? key,

? ? }

? ? ls.limiters[key] = l

? ? return l

}

// 清除過期的限流器

func (ls *Limiters) clearLimiter() {

? ? for {

? ? ? ? time.Sleep(1 * time.Minute)

? ? ? ? ls.lock.Lock()

? ? ? ? for i, i2 := range ls.limiters {

? ? ? ? ? ? //超過1分鐘

? ? ? ? ? ? if time.Now().Unix()-i2.lastGet.Unix() > 60 {

? ? ? ? ? ? ? ? delete(ls.limiters, i)

? ? ? ? ? ? }

? ? ? ? }

? ? ? ? ls.lock.Unlock()

? ? }

}

main.go結合gin框架實踐：

package main

import (

? ? "gin/component/limiter"

? ? "time"

? ? "github.com/gin-gonic/gin"

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

)

func main() {

? ? r := gin.Default()

? ? r.GET("/ceshi", Ceshi)

? ? r.Run(":8080")

}

func Ceshi(ctx *gin.Context) {

? ? l := limiter.NewLimiter(rate.Every(1*time.Second), 1, ctx.ClientIP())

? ? if !l.Allow() {

? ? ? ? ctx.JSON(400, "請求過于頻繁")

? ? ? ? return

? ? }

? ? ctx.JSON(200, "請求正常")

}

/**************************************************/

一. 限流與time/rate基礎
golang內部基于令牌桶算法提供了一個限流器time/rate位于golang.org/x/time/rate
在golang中,可以使用channel或者使用time/rate包實現并發控制, time/rate包是一個提供令牌桶算法的包，它可以實現對事件發生的速率進行限制和平滑,使用time/rate控制并發比使用channel控制并發的優點在于
time/rate只需要調用NewLimiter()函數,設置并發限制規則創建Limiter對象，然后在每個goroutine中調用Wait或者Allow方法來獲取令牌即可,簡單方便,channel需要創建一個緩沖區,然后在每個goroutine中通過向channel的緩沖區存放或獲取數據來控制并發
time/rate可以更靈活地控制并發的速率,可以根據令牌桶的容量和填充速度來調整并發的上限和平均值,而channel只能根據緩沖區的大小來控制并發的上限,不能控制平均值
time/rate可以更容易地處理并發的異常情況，例如超時或者取消:提供了WaitN、Reserve、Context等方法來支持超時或者取消的場景,而channel需要額外的邏輯來處理超時或者取消，例如使用select語句或者context包
WaitN方法和Reserve方法都可以獲取n個令牌,它們以后什么不同
WaitN方法會阻塞當前goroutine,直到獲取到n個令牌,或者超時或者取消,它接收一個context參數,用于控制超時或者取消的行為,如果獲取成功,它會返回nil,否則返回一個錯誤
Reserve方法會預定n個令牌,返回一個Reservation對象用于表示預定的結果,Reservation對象提供了Delay方法,用于獲取預定的延遲時間,以及Cancel方法用于取消預定。如果預定成功它會返回一個非nil的Reservation對象否則返回nil
一般來說如果想要同步地等待令牌,或者不關心延遲時間,你可以使用WaitN方法,如果想要異步地等待令牌,或者想要知道延遲時間,你可以使用Reserve方法
time/rate常用的API概述:
NewLimiter(r Limit, b int) *Limiter: 創建一個新的限流器，r表示每秒可以向令牌桶中產生多少令牌，b表示最大容量
Limit() Limit: 返回限流器的最大事件頻率
Burst() int: 返回限流器的最大突發大小,即一次可以消費的最大令牌數1
Allow() bool: 相當于AllowN(time.Now(), 1),表示是否可以消費一個令牌,可以返回true并消費一個令牌
AllowN(t time.Time, n int) bool: 表示是否可以消費n個令牌,可以則返回true并消費n個令牌
Reserve() Reservation: 相當于ReserveN(time.Now(), 1),表示預訂一個令牌,并返回一個Reservation對象,該對象可以用來獲取需要等待的時間或者取消預訂
ReserveN(t time.Time, n int) Reservation: 表示預訂n個令牌,并返回一個Reservation對象
Wait(ctx context.Context) (err error): 相當于WaitN(ctx, 1),表示等待直到可以消費一個令牌,或者通過ctx被取消
WaitN(ctx context.Context, n int) (err error): 表示等待直到可以消費n個令牌,或者通過ctx被取消
預訂的令牌時,返回的Reservation結構體上常用的API:
Cancel(): 取消預訂,將令牌歸還給限流器
CancelAt(t time.Time): 在指定的時間取消預訂,將令牌歸還給限流器
Delay(): 返回需要等待的時間,相當于DelayFrom(time.Now())
DelayFrom(t time.Time): 返回從指定的時間開始需要等待的時間
OK()：返回預訂是否有效,如果預訂的令牌數超過了限流器的突發大小(也就是最大個數),預定無效返回false
time/rate基礎使用示例
import (
?? ?"context"
?? ?"fmt"
?? ?"github.com/gin-gonic/gin"
?? ?"golang.org/x/time/rate"
?? ?"log"
?? ?"time"
)

func main() {
?? ?//1.初始化 limiter 每秒10個令牌，令牌桶容量為20
?? ?//第一個參數r Limit代表每秒可以向桶中產生多少令牌,Limit 實際上是 float64 的別名
?? ?//第二個參數b int代表桶的容量大小,當前為20,如果桶中有20個令牌,
?? ?//可以立即獲取20個令牌,不需要等待直接執行,也就是最大并發數
?? ?limiter := rate.NewLimiter(rate.Every(time.Millisecond*100), 20)

?? ?//2.獲取1個令牌,獲取到返回true,否則false
?? ?bo := limiter.Allow()
?? ?if bo {
?? ??? ?fmt.Println("獲取令牌成功")
?? ?}

?? ?//2.獲取指定時間內指定個數令牌,獲取到返回true,
?? ?//實際上方的Allow()內部調用的就是AllowN()
?? ?limiter.AllowN(time.Now(), 2)

?? ?//3.阻塞直到獲取足夠的令牌或者上下文取消
?? ?ctx, _ := context.WithTimeout(context.Background(), time.Second*10)
?? ?limiter.Wait(ctx)
?? ?err := limiter.WaitN(ctx, 20)
?? ?if err != nil {
?? ??? ?fmt.Println("error", err)
?? ?}

?? ?//4.可以理解為預定一個令牌,當調用Reserve后不管是否存在有效令牌都會返回一個Reservation指針對象
?? ?//接下來可以通過返回的Reservation進行指定操作
?? ?reservation := limiter.Reserve()
?? ?if 0 == reservation.Delay() {
?? ??? ?fmt.Println("獲取令牌成功")
?? ?}

?? ?//5.指定實際內預定指定個數令牌
?? ?//上面Reserve()內部實際就是調用的ReserveN()
?? ?limiter.ReserveN(time.Now(), 1)

?? ?//6.修改令牌生成速率
?? ?limiter.SetLimit(rate.Every(time.Millisecond * 100))
?? ?limiter.SetLimitAt(time.Now(), rate.Every(time.Millisecond*100))

?? ?//7.修改令牌桶大小,也就是生成令牌的最大數量限制
?? ?limiter.SetBurst(50)
?? ?limiter.SetBurstAt(time.Now(), 50)

?? ?//8.獲取限流的速率即結構體中limit的值,每秒允許處理的事件數量,即每秒處理事件的頻率
?? ?l := limiter.Limit()
?? ?fmt.Printf("每秒允許處理的事件數量,即每秒處理事件的頻率為: %v", l)
?? ?//9.返回桶的最大容量
?? ?limiter.Burst()
}

WaitN實現超時取消示例
time/rate基于WaitN方法實現超時控制:限制每秒只允許10個請求,最大允許請求數為20,當超過20個后,超過部分等待30秒,如果30秒還獲取不到則拒絕,示例中寫明中文注釋,寫明哪里是拒絕執行的,哪里是放行執行的
package main

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

func main() {
?? ?// 創建一個每秒允許10個請求，最大允許請求數為20的限速器
?? ?limiter := rate.NewLimiter(10, 20)
?? ?// 模擬100個請求
?? ?for i := 0; i < 100; i++ {
?? ??? ?// 創建一個帶有30秒超時時間的context
?? ??? ?ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
?? ??? ?defer cancel()
?? ??? ?// 等待獲取一個令牌,(如果當前獲取的令牌數超過了最大限制,或者通過context超時取消了,返回錯誤)
?? ??? ?err := limiter.WaitN(ctx, 1)
?? ??? ?if err != nil {
?? ??? ??? ?// 處理錯誤，表示拒絕執行
?? ??? ??? ?fmt.Println(i, "rejected:", err)?
?? ??? ??? ?continue
?? ??? ?}
?? ??? ?// 沒有錯誤，表示放行執行
?? ??? ?fmt.Println(i, "accepted")
?? ?}
}

time/rate基于WaitN方法實現超時控制:
限制每秒只允許10個請求,最大允許請求數為20,
當請求超過20個后,判斷當前是否有正在阻塞等待的請求,如果有判斷阻塞等待的請求數是否超過5個,如果阻塞等待請求超過5個直接拒絕請求
如果沒有阻塞等待請求,或者阻塞等待請求沒有超過5個,阻塞等待30秒,如果30秒未執行,響應超時
package main

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

func main() {
?? ?// 創建一個每秒允許10個請求，最大允許請求數為20的限速器
?? ?limiter := rate.NewLimiter(10, 20)
?? ?// 創建一個原子計數器，用于記錄阻塞等待的請求數
?? ?var waiting int64 = 0
?? ?// 模擬100個請求
?? ?for i := 0; i < 100; i++ {

?? ??? ?if atomic.LoadInt64(&waiting) > 5 {
?? ??? ??? ?// 如果阻塞等待的請求數超過5個，直接拒絕執行
?? ??? ??? ?fmt.Println(i, "rejected: too many waiting requests")?
?? ??? ??? ?continue
?? ??? ?}
?? ??? ?// 增加計數器
?? ??? ?atomic.AddInt64(&waiting, 1)
?? ??? ?
?? ??? ?// 創建一個帶有30秒超時時間的context
?? ??? ?ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
?? ??? ?defer cancel()
?? ??? ?
?? ??? ?// 嘗試獲取一個令牌，(如果當前獲取的令牌數超過了最大限制,或者通過context超時取消了,返回錯誤)
?? ??? ?err := limiter.WaitN(ctx, 1)
?? ??? ?if err != nil {
?? ??? ??? ?// 處理錯誤，表示拒絕執行
?? ??? ??? ?fmt.Println(i, "rejected:", err)
?? ??? ??? ?continue
?? ??? ?}
?? ??? ?//當獲取到令牌,減少計數器(獲取到令牌就累減計數器,總覺得有點問題)
?? ??? ?atomic.AddInt64(&waiting, -1)
?? ??? ?// 沒有錯誤，表示放行執行
?? ??? ?fmt.Println(i, "accepted")?
?? ?}
}

ReserveN預定令牌示例
在執行Limiter下的Reserve()或ReserveN()方法時會返回一個Reservation,Reservation下的方法使用示例,注意ReserveN只能預訂最大限制以內的,如果ReserveN預定數量超過了最大限制
func test() {
?? ?//1.初始化 limiter 每秒10個令牌，令牌桶容量為20
?? ?limiter := rate.NewLimiter(rate.Every(time.Millisecond*100), 20)

?? ?//2.可以理解為預定一個令牌,當調用Reserve后不管是否存在有效令牌都會返回一個Reservation指針對象
?? ?//接下來可以通過返回的Reservation進行指定操作
?? ?reservation := limiter.Reserve()
?? ?if !reservation .OK() {
?? ??? ?// 預定失敗，表示超過了最大突發數
?? ?}

?? ?//獲取到Reservation后

?? ?//3.Delay():如果ReserveN預訂超過最大限制,Delay()方法會返回一個非零的值
?? ?//返回需要阻塞等待多長時間才能拿到令牌,如果為0說明不用阻塞等待
?? ?if 0 == reservation.Delay() {
?? ??? ?fmt.Println("獲取令牌成功")
?? ?}

?? ?//4.上面Delay()內部就是調用的該方法,如果返回0，表示有足夠的令牌,
?? ?//如果返回InfDuration，表示到截至時間時仍然沒有足夠的令牌
?? ?reservation.DelayFrom(time.Now())

?? ?//5.返回限流器limiter是否可以在最大等待時間內提供請求數量的令牌。
?? ?//如果Ok為false，則Delay返回InfDuration，Cancel不執行任何操作
?? ?if reservation.OK() {
?? ??? ?fmt.Println("獲取令牌成功")
?? ?}

?? ?//6.用于取消預約令牌操作,如果有需要還原的令牌,
?? ?//則將需要還原的令牌重新放入到令牌桶中,注意并不是無腦還原
?? ?reservation.Cancel()
?? ?//上方的Cancel()內部就是調用的CancelAt()這個方法
?? ?reservation.CancelAt(time.Now())
}

使用示例
需求: 根據調用服務或者接口限流
指定服務或接口有限流要求,獲取到限流規則
編寫限流邏輯,創建限流結構,限流容器,獲取限流limit方法
提供限流中間件,獲取到請求后獲取限流limit,如果超出閾值熔斷
限流邏輯
import (
?? ?"golang.org/x/time/rate"
?? ?"sync"
)

// 1.針對服務(或者針對接口)限流,封裝限流結構
type FlowLimiterItem struct {
?? ?ServiceName string ? ? ? ?//需要限流的服務或接口標識
?? ?Limter ? ? ?*rate.Limiter //限流limit
}

// 1.封裝限流容器(不同服務或者不同接口的限流上下文存儲到該容器中)
type FlowLimiter struct {
?? ?FlowLmiterMap ? map[string]*FlowLimiterItem
?? ?FlowLmiterSlice []*FlowLimiterItem
?? ?Locker ? ? ? ? ?sync.RWMutex
}

// 3.提供初始化限流容器函數
func NewFlowLimiter() *FlowLimiter {
?? ?return &FlowLimiter{
?? ??? ?FlowLmiterMap: ? map[string]*FlowLimiterItem{},
?? ??? ?FlowLmiterSlice: []*FlowLimiterItem{},
?? ??? ?Locker: ? ? ? ? ?sync.RWMutex{},
?? ?}
}

//4.服務啟動時初始化限流容器
var FlowLimiterHandler *FlowLimiter

func init() {
?? ?FlowLimiterHandler = NewFlowLimiter()
}

//5.獲取指定服務或接口限流limit方法
//serverName:服務或接口標識
//qps:該服務或接口限流qps
func (counter *FlowLimiter) GetLimiter(serverName string, qps float64) (*rate.Limiter, error) {

?? ?//1.通過服務或接口標識在限流容器中獲取
?? ?for _, item := range counter.FlowLmiterSlice {
?? ??? ?if item.ServiceName == serverName {
?? ??? ??? ?return item.Limter, nil
?? ??? ?}
?? ?}

?? ?//2.如果容器中不存在說明第一次執行,需要創建限流limit
?? ?newLimiter := rate.NewLimiter(rate.Limit(qps), int(qps*3))
?? ?//封裝限流結構添加到限流容器中
?? ?item := &FlowLimiterItem{
?? ??? ?ServiceName: serverName,
?? ??? ?Limter: ? ? ?newLimiter,
?? ?}
?? ?counter.FlowLmiterSlice = append(counter.FlowLmiterSlice, item)
?? ?counter.Locker.Lock()
?? ?defer counter.Locker.Unlock()
?? ?counter.FlowLmiterMap[serverName] = item
?? ?return newLimiter, nil
}

編寫限流中間件
import (
?? ?"fmt"
?? ?"github.com/e421083458/go_gateway/dao"
?? ?"github.com/e421083458/go_gateway/middleware"
?? ?"github.com/e421083458/go_gateway/public"
?? ?"github.com/gin-gonic/gin"
?? ?"github.com/pkg/errors"
)

func HTTPFlowLimitMiddleware() gin.HandlerFunc {
?? ?return func(c *gin.Context) {
?? ??? ?//1.獲取請求參數
?? ??? ?serverInterface, ok := c.Get("service")
?? ??? ?if !ok {
?? ??? ??? ?middleware.ResponseError(c, 2001, errors.New("service not found"))
?? ??? ??? ?c.Abort()
?? ??? ??? ?return
?? ??? ?}
?? ??? ?//2.根據請求數據拿查詢限流配置
?? ??? ?serviceDetail := serverInterface.(*dao.ServiceDetail)

?? ??? ?//3.根據配置判斷限流規則
?? ??? ?if serviceDetail.AccessControl.ServiceFlowLimit != 0 {

?? ??? ??? ?//4.獲取限流limit
?? ??? ??? ?serviceLimiter, err := public.FlowLimiterHandler.GetLimiter(
?? ??? ??? ??? ?public.FlowServicePrefix+serviceDetail.Info.ServiceName,
?? ??? ??? ??? ?float64(serviceDetail.AccessControl.ServiceFlowLimit))
?? ??? ??? ?if err != nil {
?? ??? ??? ??? ?middleware.ResponseError(c, 5001, err)
?? ??? ??? ??? ?c.Abort()
?? ??? ??? ??? ?return
?? ??? ??? ?}

?? ??? ??? ?//5.判斷限流是否超過閾值,超過返回false,進入if進行異常響應
?? ??? ??? ?if !serviceLimiter.Allow() {
?? ??? ??? ??? ?middleware.ResponseError(c, 5002, errors.New(fmt.Sprintf("service flow limit %v", serviceDetail.AccessControl.ServiceFlowLimit)))
?? ??? ??? ??? ?c.Abort()
?? ??? ??? ??? ?return
?? ??? ??? ?}
?? ??? ?}

?? ??? ?if serviceDetail.AccessControl.ClientIPFlowLimit > 0 {
?? ??? ??? ?clientLimiter, err := public.FlowLimiterHandler.GetLimiter(
?? ??? ??? ??? ?public.FlowServicePrefix+serviceDetail.Info.ServiceName+"_"+c.ClientIP(),
?? ??? ??? ??? ?float64(serviceDetail.AccessControl.ClientIPFlowLimit))
?? ??? ??? ?if err != nil {
?? ??? ??? ??? ?middleware.ResponseError(c, 5003, err)
?? ??? ??? ??? ?c.Abort()
?? ??? ??? ??? ?return
?? ??? ??? ?}
?? ??? ??? ?if !clientLimiter.Allow() {
?? ??? ??? ??? ?middleware.ResponseError(c, 5002, errors.New(fmt.Sprintf("%v flow limit %v", c.ClientIP(), serviceDetail.AccessControl.ClientIPFlowLimit)))
?? ??? ??? ??? ?c.Abort()
?? ??? ??? ??? ?return
?? ??? ??? ?}
?? ??? ?}

?? ??? ?//6.中間件放行
?? ??? ?c.Next()
?? ?}
}

該方式也可以再提取出一個針對租戶限流的中間件
import (
?? ?"fmt"
?? ?"github.com/e421083458/go_gateway/dao"
?? ?"github.com/e421083458/go_gateway/middleware"
?? ?"github.com/e421083458/go_gateway/public"
?? ?"github.com/gin-gonic/gin"
?? ?"github.com/pkg/errors"
)

func HTTPJwtFlowLimitMiddleware() gin.HandlerFunc {
?? ?return func(c *gin.Context) {
?? ??? ?//1.獲取請求
?? ??? ?appInterface, ok := c.Get("app")
?? ??? ?if !ok {
?? ??? ??? ?c.Next()
?? ??? ??? ?return
?? ??? ?}
?? ??? ?//2.通過請求查詢該app用戶下的限流配置
?? ??? ?appInfo := appInterface.(*dao.App)
?? ??? ?//3.判斷是否開啟限流
?? ??? ?if appInfo.Qps > 0 {
?? ??? ??? ?clientLimiter, err := public.FlowLimiterHandler.GetLimiter(
?? ??? ??? ??? ?public.FlowAppPrefix+appInfo.AppID+"_"+c.ClientIP(),
?? ??? ??? ??? ?float64(appInfo.Qps))
?? ??? ??? ?if err != nil {
?? ??? ??? ??? ?middleware.ResponseError(c, 5001, err)
?? ??? ??? ??? ?c.Abort()
?? ??? ??? ??? ?return
?? ??? ??? ?}
?? ??? ??? ?//4.判斷是否到達限流閾值
?? ??? ??? ?if !clientLimiter.Allow() {
?? ??? ??? ??? ?middleware.ResponseError(c, 5002, errors.New(fmt.Sprintf("%v flow limit %v", c.ClientIP(), appInfo.Qps)))
?? ??? ??? ??? ?c.Abort()
?? ??? ??? ??? ?return
?? ??? ??? ?}
?? ??? ?}
?? ??? ?c.Next()
?? ?}
}

二. time/rate 底層原理相關
Limiter 與 Reservation 結構
在調用NewLimiter()創建限流器時會返回一個Limiter結構體變量,查看內部組成:
type Limiter struct {
?? ?//每秒允許處理的事件數量,即每秒處理事件的頻率
?? ?limit Limit ?
?? ?//令牌桶的最大數量,如果burst為0,則除非limit == Inf,否則不允許處理任何事件
?? ?burst int ??

?? ?mu ? ? sync.Mutex
?? ?//令牌桶中可用的令牌數量
?? ?tokens float64 ?
?? ?//記錄上次limiter的tokens被更新的時間
?? ?last time.Time
?? ?//lastEvent記錄速率受限制(桶中沒有令牌)的時間點,該時間點可能是過去的
?? ?//也可能是將來的(Reservation預定的結束時間點)
?? ?//如果沒有預約令牌的話,該時間等于last，是過去的
?? ?//如果有預約令牌的話,該時間等于最新的預約的截至時間
?? ?lastEvent time.Time ?
}

當調用Reserve()預定令牌時,會返回一個Reservation結構體變量
type Reservation struct {
?? ?//到截止時間是否可以獲取足夠的令牌
?? ?ok ? ? ? ?bool?
?? ?lim ? ? ? *Limiter
?? ?//需要獲取的令牌數量
?? ?tokens ? ?int ??
?? ?//需要等待的時間點(本次預約需要等待到的指定時間點才有足夠預約的令牌)
?? ?timeToAct time.Time ?
?? ?limit Limit
}

消費令牌底層原理
在rate中默認提供了Allow()消費一個令牌,Wait()阻塞等待消費一個令牌, Reserve()預定一個令牌,實際底層對應調用的是WaitN(), AllowN(), ReserveN()消費指定個數令牌的方法
所有消費令牌的方法內部都會調用reserveN和advance方法,reserveN可以理解為預約令牌的邏輯,由于可以預約,當桶中令牌不夠時,預定過后桶中令牌有可能為負數
以Allow()消費一個令牌為例,內部會調用AllowN(), AllowN內部會調用reserveN(),reserveN內部會調用advance()
func (lim *Limiter) Allow() bool {
?? ?return lim.AllowN(time.Now(), 1)
}

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

1. reserveN()方法邏輯
reserveN是 AllowN, ReserveN及 WaitN的輔助方法,主要用于判斷在maxFutureReserve指定時間內是否有足夠的令牌
// @param n 要消費的token數量
// @param maxFutureReserve 愿意等待的最長時間
func (lim *Limiter) reserveN(t time.Time, n int, maxFutureReserve time.Duration) Reservation {
?? ?lim.mu.Lock() //加鎖，保證數據的一致性
?? ?defer lim.mu.Unlock() //解鎖，釋放資源

?? ?if lim.limit == Inf { //如果速率限制是無限的，表示不需要限流
?? ??? ?return Reservation{ //返回一個預約結果，包括是否允許、令牌數、執行時間等信息
?? ??? ??? ?ok: ? ? ? ?true, //允許通過
?? ??? ??? ?lim: ? ? ? lim, //關聯的限流器
?? ??? ??? ?tokens: ? ?n, //請求的令牌數
?? ??? ??? ?timeToAct: t, //執行時間為當前時間
?? ??? ?}
?? ?} else if lim.limit == 0 { //如果速率限制是零，表示不允許任何事件
?? ??? ?var ok bool?
?? ??? ?if lim.burst >= n { //如果桶中有足夠的令牌，表示可以通過
?? ??? ??? ?ok = true?
?? ??? ??? ?lim.burst -= n //更新桶中的令牌數，減去請求的令牌數
?? ??? ?}
?? ??? ?return Reservation{ //返回一個預約結果，包括是否允許、令牌數、執行時間等信息
?? ??? ??? ?ok: ? ? ? ?ok, //是否允許通過
?? ??? ??? ?lim: ? ? ? lim, //關聯的限流器
?? ??? ??? ?tokens: ? ?lim.burst, //桶中剩余的令牌數
?? ??? ??? ?timeToAct: t, //執行時間為當前時間
?? ??? ?}
?? ?}

?? ?t, tokens := lim.advance(t) //根據當前時間和上次更新時間計算桶中剩余的令牌數

?? ?tokens -= float64(n) //計算請求后桶中剩余的令牌數，可能為負數

?? ?var waitDuration time.Duration?
?? ?if tokens < 0 { //如果剩余的令牌數小于零，表示需要等待一段時間才能通過
?? ??? ?waitDuration = lim.limit.durationFromTokens(-tokens) //根據速率限制計算需要等待的時間間隔
?? ?}

?? ?//判斷是否允許通過，需要滿足兩個條件：請求的令牌數不超過桶的容量，等待的時間間隔不超過最大允許的未來預約時間
?? ?ok := n <= lim.burst && waitDuration <= maxFutureReserve?

?? ?r := Reservation{ //準備一個預約結果，包括是否允許、關聯的限流器、速率限制等信息
?? ??? ?ok: ? ?ok,?
?? ??? ?lim: ? lim,?
?? ??? ?limit: lim.limit,?
?? ?}
?? ?if ok { //如果允許通過，還需要設置以下信息：
?? ??? ?r.tokens = n //請求的令牌數
?? ??? ?r.timeToAct = t.Add(waitDuration) //執行時間為當前時間加上等待時間間隔

?? ??? ?// Update state
?? ??? ?lim.last = t //更新限流器中的上次更新時間為當前時間
?? ??? ?lim.tokens = tokens //更新限流器中的剩余令牌數為請求后的值
?? ??? ?lim.lastEvent = r.timeToAct //更新限流器中的最近事件時間為執行時間
?? ?}
?? ?return r //返回預約結果
}

對reserveN的流程總結
調用NewLimiter()創建限流器時會返回一個Limiter結構體變量,內部存在一個limit每秒允許處理的事件數量,與burst最大通過數量屬性
在調用reserveN()消費令牌時,首先會加鎖,判斷limit是否等于MaxFloat64,如果是說明無限的速率限制桶中一直擁有足夠的令牌,直接返回true
如果limit等于0,判斷當前獲取的令牌數量是否超過了burst最大并發限制,如果超過了返回false,沒超過返回true
如果limit不等于MaxFloat64并且不等于0,調用advance()計算當前可以使用的令牌數量,也就是(上次剩余令牌數+上次消費令牌時間到當前時間所生成的令牌,如果大于最大并發限制,的更新為最大并發限制)
如果advance()拿到當前可消費的令牌數量小于0,說明超過并發限制,調用durationFromTokens()計算等待生成令牌時間
封裝Reservation,如果當前要消費的令牌數量小于允許可消費的令牌數量則Reservation中的ok為true表示允許消費,否則為false
最后解鎖,返回預約結果
2. advance()根據當前時間和上次更新時間計算桶中剩余的令牌數邏輯
該方法的作用是更新令牌桶的狀態,計算出令牌桶未更新的時間(elapsed),根據elapsed算出需要向桶中加入的令牌數delta，然后算出桶中可用的令牌數newTokens
func (lim *Limiter) advance(t time.Time) (newT time.Time, newTokens float64) {
?? ?last := lim.last //獲取限流器中的上次更新時間
?? ?if t.Before(last) { //如果當前時間早于上次更新時間，表示時間回退了
?? ??? ?last = t //使用當前時間作為上次更新時間
?? ?}

?? ?elapsed := t.Sub(last) //計算當前時間和上次更新時間的差值，單位納秒
?? ?delta := lim.limit.tokensFromDuration(elapsed) //根據速率限制計算在這段時間內應該增加的令牌數
?? ?tokens := lim.tokens + delta //計算桶中新的令牌數，等于原來的令牌數加上增加的令牌數
?? ?if burst := float64(lim.burst); tokens > burst { //如果新的令牌數超過了桶的容量
?? ??? ?tokens = burst //將新的令牌數設置為桶的容量，即不能超過最大值
?? ?}
?? ?return t, tokens //返回當前時間和新的令牌數
}

3. durationFromTokens()根據令牌數量tokens計算出產生該數量的令牌需要的時長
durationFromTokens()計算出生成N 個新的 Token 一共需要多久
func (limit Limit) durationFromTokens(tokens float64) time.Duration {
?? ?if limit <= 0 { //如果速率限制小于等于零，表示不需要任何時間
?? ??? ?return InfDuration //返回一個無限大的時間間隔
?? ?}
?? ?seconds := tokens / float64(limit) //計算生成令牌數所需的秒數，等于令牌數除以速率限制
?? ?return time.Duration(float64(time.Second) * seconds) //返回對應的時間間隔，單位納秒
}

4. tokensFromDuration()獲取指定期間內產生的令牌數量
tokensFromDuration()給定一段時長,計算這段時間一共可以生成多少個 Token
func (limit Limit) tokensFromDuration(d time.Duration) float64 {
?? ?if limit <= 0 { //如果速率限制小于等于零，表示不生成任何令牌
?? ??? ?return 0
?? ?}
?? ? //返回時間間隔的秒數乘以速率限制，即在這段時間內生成的令牌數
?? ?return d.Seconds() * float64(limit)
}

Wait 阻塞獲取令牌
在Wait阻塞消費令牌函數中,首先會封裝一個返回定時器通道的函數,然后調用Limiter的wait()方法, 查看該方法:
首先獲取到當前限速器配置的最大并發數burst 與limit限流速率,如果當前一次性申請的令牌數超過最大并發數,并且限流速率不是MaxFloat64報錯
通過select-case監聽context的Done取消,如果取消了返回異常
將Context取消時間,設置為等待時間
通過當前時間, 當前消費令牌數量,通過Context取消時間設置的等待時間,調用reserveN()進行預訂
在reserveN()預訂令牌方法中,會根據傳入的等待時間計算這個時間段內是否存在或可以生產出指定數量的令牌,并且會返回生產這些令牌所需要的時間
如果存在或者可以生產出需要的令牌reserveN()返回的Reservation中ok為true,調用DelayFrom()方法計算獲取令牌需要的延遲時間,DelayFrom()返回0說明不需要等待直接返回
DelayFrom()如果返回大于0,執行Wait()函數中封裝的返回定時器通道的函數,拿到定時器通道,通過select-case監聽Context的Done取消消息和這個定時器通道,當定時器通道返回說明令牌生成完畢函數返回,Wait阻塞開始向下執行
// WaitN 會阻塞當前 goroutine 直到可以獲取 n 個令牌，或者 ctx 被取消
// 如果 n 超過了 limiter 的 burst 值，并且 limit 不是 Inf，WaitN 會返回一個錯誤
// 如果 ctx 已經被取消，WaitN 也會返回一個錯誤
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
?? ?// 這是真正的定時器生成器
?? ?// newTimer 是一個函數，它接受一個時間間隔 d 作為參數，
?? ?//返回一個定時器的通道，一個停止定時器的函數，和一個空函數(只在測試時有作用)
?? ?newTimer := func(d time.Duration) (<-chan time.Time, func() bool, func()) {
?? ??? ?// 創建一個定時器，它會在 d 時間后發送當前時間到 timer.C 通道
?? ??? ?timer := time.NewTimer(d)?
?? ??? ?// 返回定時器的通道，停止定時器的函數，和空函數
?? ??? ?return timer.C, timer.Stop, func() {}?
?? ?}
?? ?// 調用 lim.wait 方法，傳入 ctx, n, 當前時間，和 newTimer 函數
?? ?return lim.wait(ctx, n, time.Now(), newTimer)?
}

// wait 是 WaitN 的內部實現
func (lim *Limiter) wait(ctx context.Context, n int, t time.Time, newTimer func(d time.Duration) (<-chan time.Time, func() bool, func())) error {
?? ?lim.mu.Lock()
?? ?burst := lim.burst // 獲取最大并發限制
?? ?limit := lim.limit // 獲取限流速率
?? ?lim.mu.Unlock()

?? ?if n > burst && limit != Inf {
?? ??? ?// 如果 n 超過了 burst 值，并且 limit 不是 Inf，返回一個錯誤
?? ??? ?return fmt.Errorf("rate: Wait(n=%d) exceeds limiter's burst %d", n, burst)?
?? ?}
?? ?// 檢查 ctx 是否已經被取消
?? ?select {
?? ?case <-ctx.Done():
?? ??? ?return ctx.Err() // 如果 ctx 已經被取消，返回一個錯誤
?? ?default:
?? ?}
?? ?// 確定等待限制
?? ?waitLimit := InfDuration // 初始化等待限制為無窮大
?? ?if deadline, ok := ctx.Deadline(); ok {
?? ??? ?// 如果 ctx 有截止時間，等待限制為截止時間減去當前時間
?? ??? ?waitLimit = deadline.Sub(t)?
?? ?}
?? ?// 預約
?? ?// 調用 lim.reserveN 方法，傳入當前時間t，n，和等待限制，返回一個預約對象 r
?? ?r := lim.reserveN(t, n, waitLimit)?
?? ?if !r.ok {
?? ??? ?// 如果 r 不 ok，說明預約失敗，返回一個錯誤
?? ??? ?return fmt.Errorf("rate: Wait(n=%d) would exceed context deadline", n)?
?? ?}
?? ?//判斷是否需要等待
?? ?delay := r.DelayFrom(t) // 獲取 r 的延遲時間
?? ?if delay == 0 {
?? ??? ?return nil // 如果延遲時間為 0直接返回 nil
?? ?}
?? ?// 調用 newTimer 函數，傳入延遲時間，返回一個定時器的通道，
?? ?//一個停止定時器的函數，和一個空函數（只在測試時有作用）
?? ?ch, stop, advance := newTimer(delay)?
?? ?defer stop() // 延遲執行停止定時器的函數
?? ?advance() // 只在測試時有作用
?? ?select {
?? ?case <-ch:
?? ??? ?// 如果定時器的通道收到消息，說明延遲時間到了，我們可以繼續，返回 nil
?? ??? ?return nil?
?? ?case <-ctx.Done():
?? ??? ?// ctx 被取消了,取消預約
?? ??? ?r.Cancel()?
?? ??? ?return ctx.Err() // 返回 ctx 的錯誤
?? ?}
}

// DelayFrom 返回 r 的延遲時間，即從 t 到 r.timeToAct 的時間間隔。
// 如果 r 不 ok，說明預約失敗，返回無窮大的延遲時間。
// 如果 r.timeToAct 在 t 之前，說明預約已經可以執行，返回 0 的延遲時間。
func (r *Reservation) DelayFrom(t time.Time) time.Duration {
?? ?if !r.ok {
?? ??? ?return InfDuration // 如果 r 不 ok，返回無窮大的延遲時間
?? ?}
?? ?delay := r.timeToAct.Sub(t) // 計算 r.timeToAct 和 t 的時間差，賦值給 delay
?? ?if delay < 0 {
?? ??? ?return 0 // 如果 delay 小于 0，說明 r.timeToAct 在 t 之前，返回 0 的延遲時間
?? ?}
?? ?return delay // 否則，返回 delay
}

CancelAt()取消令牌消費操作
func (r *Reservation) Cancel() {
?? ?r.CancelAt(time.Now())
?? ?return
}
?
func (r *Reservation) CancelAt(t time.Time) {
?? ?//如果預約結果不允許通過，表示沒有消耗令牌，無需取消
?? ?if !r.ok {?
?? ??? ?return
?? ?}

?? ?r.lim.mu.Lock() //加鎖，保證數據的一致性
?? ?defer r.lim.mu.Unlock() //解鎖，釋放資源

?? ?//如果速率限制是無限的，或者請求的令牌數是零，或者執行時間早于取消時間，表示無需取消
?? ?if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(t) {?
?? ??? ?return
?? ?}

?? ?//計算需要恢復的令牌數，等于請求的令牌數減去在預約結果之后被預約的令牌數
?? ?//這里的r.lim.lastEvent可能是本次Reservation的結束時間
?? ?//也可能是后來的Reservation的結束時間
?? ?//所以要把本次結束時間點(r.timeToAct)之后產生的令牌數減去
?? ?restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))?
?? ?if restoreTokens <= 0 { //如果需要恢復的令牌數小于等于零，表示無需取消
?? ??? ?return
?? ?}
?? ?t, tokens := r.lim.advance(t) //根據當前時間和上次更新時間計算桶中剩余的令牌數

?? ?tokens += restoreTokens //計算桶中新的令牌數，等于原來的令牌數加上恢復的令牌數
?? ?if burst := float64(r.lim.burst); tokens > burst { //如果新的令牌數超過了桶的容量
?? ??? ?tokens = burst //將新的令牌數設置為桶的容量，即不能超過最大值
?? ?}
?? ?// update state
?? ?r.lim.last = t //更新限流器中的上次更新時間為當前時間
?? ?r.lim.tokens = tokens //更新限流器中的剩余令牌數為新的令牌數
?? ?//如果預約結果的執行時間等于限流器中的最近事件時間，表示需要調整最近事件時間
?? ?if r.timeToAct == r.lim.lastEvent {?
?? ??? ?//計算預約結果之前的最近事件時間，等于執行時間減去生成請求令牌數所需的時間間隔
?? ??? ?prevEvent := r.timeToAct.Add(r.limit.durationFromTokens(float64(-r.tokens)))?
?? ??? ?if !prevEvent.Before(t) { //如果預約結果之前的最近事件時間不早于當前時間，表示有效
?? ??? ??? ?r.lim.lastEvent = prevEvent //更新限流器中的最近事件時間為預約結果之前的最近事件時間
?? ??? ?}
?? ?}
}

可以調用Cancel()取消令牌消費操作,該函數中會調用CancelAt(),首先要了解Reservation中的幾個字段
r.tokens指的是本次消費的token數，
r.timeToAcr指的是Token桶可以滿足本次消費數目的時刻，也就是消費的時刻+等待的時長
r.lim.lastEvent指的是最近一次消費的timeToAct的值
在CancelAt()中最重要的邏輯:通過r.limit.tokensFromDuration方法得出從該次消費到當前時間一共又消費了多少Token數目,然后"r.tokens本次消費的令牌數" 減去 "從該次消費到當前時間一共又消費的令牌數"得出要歸還的令牌
restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))，
1
然后更新Reservation中的上次token更新時間,剩余令牌數等信息
三. 總結
復習限流算法
先說一下幾個限流相關算法的優缺點
計數器: 只需要維護一個計數器變量,實現簡單,性能消耗較小,缺點無法控制速率,無法解決瞬時高并發問題
令牌桶:
固定速率,向桶中添加令牌,桶容量是有限的,接收到請求后首先獲取桶中的令牌進行消費
缺點: 需要維護令牌生成速率與桶容量實現稍微有點復雜,通過桶容量一定程度上解決了瞬時并發問題,但是沒有徹底解決
漏桶
可以以任意速率流入水滴到漏桶中,但是按照固定速率流出水滴, 如果桶是空的，則不需流出水滴,；如果流入水滴超出了桶的容量，則流入的水滴溢出了執行服務降級
令牌桶是按照固定速率往桶中添加令牌,請求是否被處理需要看桶中令牌是否足夠,當令牌數減為零時則拒絕新的請求；漏桶則是按照常量固定速率流出請求,請求到達后端,流入請求速率任意請求有客戶端發送到桶中,當流入的請求數累積到漏桶容量時,則新流入的請求被拒絕
缺點: 與令牌桶相同
滑動窗口
滑動窗口出現的原因: 假設通過計數器實現限流,限制每秒鐘最高允許10個請求通過,每請求一次計數器+1,當請求超過10,并且與第一次請求的時間間隔不超過一秒鐘,說明請求過多,服務熔斷降級,如果與第一次請求間隔超過一秒鐘,說明還在范圍內重置計數器(思考問題臨界問題:假設在第1秒時接收到9個請求,都在執行中,到2秒計數器重置進入新的計算階段又進來9個請求,當前實際就承載了18個請求,超過閾值)
滑動窗口算法計數器: 解決傳統計數器臨界問題: 假設每分鐘允許向后臺請求60次,可以將這60次請求分為6份,每秒鐘允許請求10次,每個時間段都有自己的計數器,記錄這個時間段內發生的請求數量,通過滑動進行判斷,每過一個時間段,就把最早的時間段和它的計數器刪掉,然后加入一個新的時間段和計數器,進而解決臨界問題
time/rate 總結
time/rate 基于令牌桶算法實現的限流組件,允許一定程度的突發,同時保證了請求的平均速率,內部基于鎖,channel,保證了并發安全,犧牲了一點性能但是保證了令牌的精確性
了解time/rate限流首先要了解Limiter與Reservation兩個結構體,在調用NewLimiter()函數創建限流器時會返回一個Limiter 結構體變量,內部有一個limit 屬性表示每秒處理的頻率也就是限流速率, burst 最大并發數,mu 鎖, tokens 當前令牌桶中可用的令牌數…
在實現限流時底層都會調用到一個reserveN()函數,會根據傳入的等待時間,計算這段時間內能否生成需要的令牌數,封裝Reservation,在Reservation 中存在,tokens本次消費的令牌數,ok 到截止時間是否能夠生成指定的令牌數屬性,timeToAct 獲取指定令牌需要等待的時間屬性
消費令牌原理:
rate中提供了Allow()消費一個令牌,Wait()阻塞等待消費一個令牌, Reserve()預定一個令牌,內部實際都會調用到調用reserveN和advance方法,reserveN()用于判斷在指定時間內是否有足夠的令牌
首先判斷當前消費的令牌數是否超過了最大限制,如果超過了,直接報錯,判斷limit限流速率如果不等于MaxFloat64并且不等于0,調用advance()計算當前可以使用的令牌數量,也就是(上次剩余令牌數+上次消費令牌時間到當前時間所生成的令牌,如果大于最大并發限制,的更新為最大并發限制)
如果advance()拿到當前可消費的令牌數量小于0,說明超過并發限制,調用durationFromTokens()計算等待生成令牌時間
封裝Reservation,如果當前要消費的令牌數量小于允許可消費的令牌數量則Reservation中的ok為true表示允許消費,否則為false,最后解鎖,返回預約結果
Wait 阻塞獲取令牌原理,在Wait阻塞消費令牌函數中,首先會封裝一個返回定時器通道的函數,然后調用Limiter的wait()方法,
在wait()方法中,會通過select-case監聽context的Done取消,如果取消了返回異常,并獲取context的取消時間,作為等待時間調用reserveN()進行預訂
在reserveN()預訂令牌方法中,會根據傳入的等待時間計算這個時間段內是否存在或可以生產出指定數量的令牌,并且會返回生產這些令牌所需要的時間
如果存在或指定等待時間內可以生產出需要的令牌reserveN()返回的Reservation中ok為true,調用DelayFrom()方法計算獲取令牌需要的等待時間,DelayFrom()返回0說明不需要等待直接返回,大于0,執行Wait()函數中封裝的返回定時器通道的函數,拿到定時器通道,通過select-case監聽Context的Done取消消息和這個定時器通道,當定時器通道返回說明令牌生成完畢函數返回,Wait阻塞開始向下執行
Cancel()取消令牌消費操作: Cancel()函數內部會調用CancelAt(),該函數中最重要的邏輯:通過r.limit.tokensFromDuration方法得出從該次消費到當前時間一共又消費了多少Token數目,然后"r.tokens本次消費的令牌數" 減去 "從該次消費到當前時間一共又消費的令牌數"得出要歸還的令牌,然后更新Reservation中的上次token更新時間,剩余令牌數等信息