go 限速与限流

2020-09-30 本文已影响0人 Best博客

限速方式

漏桶算法：讲究的是服务器匀速的去处理并发请求，但... 为达到目的居然采用sleep了。简单来说服务器匀速处理请求，超过桶容量会被舍弃
令牌桶算法：拿到令牌的请求被处理，否则被舍弃。在桶里面的令牌被拿光了的时候，此时就是一边生产令牌一边消耗令牌了，这种场景下也匀速了。存峰值，峰值为桶容量 + 消耗此容量所需时间产生的新token。

你需要搞清楚

熔断与限速与限流是比较常用的服务降级方案，属于主动触发。
服务熔断：打比方你对外提供的服务又5个子服务组成，其中2个是核心服务，3个是非核心服务，这个时候或许你可以考虑下当服务过载的时候，自动熔断掉那3个非核心的服务，同时触发报警，为扩容争取时间。
限速。但如果去掉非核心服务后负载还下不来，那可能真的是因为核心服务并发量突然太高了，你只能扛10万现在来了100万，为了避免一下把你服务打死，你就必须通过限速策略，限速并不能保证100万请求全正常，但能保证服务器一直以近视于10万的qps正常吞吐（当然并发实在太大没到应用层可能就挂了，这里暂不考虑那种），也就是丢了90万。。。
限速是限制流量流入的速度，才不会管你服务器 hang了没。打比方你服务器正常情况下，10万qps，现在来了100万流量并将持续1小时，由于限速的作用并没有一下打死你的服务器，然后你的服务器任然以10万的qps提供服务，过30分钟后突然db出现慢查询了，请注意这个时候相当于你服务器每秒qps没有10万了，但是限速限制的还是10万，这个时候很可能你服务器即将gg。。。
限流着重是控制并发的最大流量。

基于 tollbooth 实现

如果要讲究开箱机即用，用这个开源组件去做http限速你只要按着demo稍微配置下。

可以配置只针对GET或POST 类型请求做限制
可以配置只针对ip 做限制
可以配置只针对请求头中带有某种特定标识的请求做限制（不重要的服务熔断可以采用它）
文档中有gin，echo等http的使用该组件的文档
可以开发成中间件
可以做到定时清理计数~
可以设置丢弃返回值
......

package main

import (
    "github.com/didip/tollbooth"
    "net/http"
)

func HelloHandler(w http.ResponseWriter, req *http.Request) {
    w.Write([]byte("Hello, World!"))
}

func main() {
    // Create a request limiter per handler.
    http.Handle("/", tollbooth.LimitFuncHandler(tollbooth.NewLimiter(1, nil), HelloHandler))
    http.ListenAndServe(":12345", nil)
}

探索 golang.org/x/time/rate

令牌桶这个算法

image.png
精简版：一个gorontinue定时往里面塞，所有的请求想要被响应必须先去channel取token，没取到的丢弃。
但感觉golang.org/x/time/rate 实现方式巧。
1.它并没有向图里面那样往 channel写token，而是通过一个基于纳秒的算法表达出了令牌桶的这种效果

它用的事lock 方式去修改全局变量各计数的属性值的，通过计数去控制流量流入速度
任何请求过来都是基于纳秒级别的去计算当前的流量流入速度是否超标，超标则舍去。
算法核心。拿到当前允许被消耗token计数，根据配置进去的消耗速度计算消耗掉该值的时间，以该时间通过配置的速度求出能够新产生的token数量，2者相加得到当前实时的token数量，自减1，其结果如果小于等于0表示拒绝。
如此它的真实放行的速度是紧紧结合你当前实际运行的效果得出来的，最峰值是你当前一个请求没有，突然来了100万流量。
这个限速不光光可以用来限制处理请求的速度，也可以用来限制你调用别个服务的速度，也就是你可以用来处理任何需要控制其执行速度的场景（指的是函数被触发的速度）
7.核心巧代码：

waitDuration = lim.limit.durationFromTokens(-tokens)  //通过tokens数量算出消耗它需要的时间

 lim.limit.tokensFromDuration(elapsed) //通过纳秒级别时间算出能够生产的 tokens数量

demo


package main

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

const (
    speed = 1  //每秒执行的次数
    capacity = 10 //桶的容量大小
)

var gameScene = rate.NewLimiter(speed , capacity )

func main() {

    for i:=0;i<100;i++{
        k :=i
        if isGameSceneAllow(){
            fmt.Println("我是被接受的请求",time.Now().Unix(),k)
        }
    }


    //9秒钟sleep，忽略代码执行时间，那么将会产生9个
    time.Sleep(time.Second * 9)


    //以下打印9个，则证明限速起作用了
    for i:=0;i<100;i++{
        k :=i
        if isGameSceneAllow(){
            fmt.Println("我是被接受的请求2222",time.Now().Unix(),k)
        }
    }

    time.Sleep(time.Second * 9)

}

func isGameSceneAllow()(b bool){
    if gameScene.Allow() == false {
        return
    }
    b =true
    return
}

限流

着重点是去限制你的服务器并发处理请求的能力。打比方你的服务器最多同时处理1万个请求，它的出现就是同时处理1万个请求，请求处理完毕资源就会被释放，就可以让新的流量进入。

package main

import (
    "log"
    "net/http"
    "text/template"
    "time"

    "github.com/julienschmidt/httprouter"
)

type middleWareHandler struct {
    r *httprouter.Router
    l *ConnLimiter
}

//NewMiddleWareHandler ...
func NewMiddleWareHandler(r *httprouter.Router, cc int) http.Handler {
    m := middleWareHandler{}
    m.r = r
    m.l = NewConnLimiter(cc)
    return m
}

func (m middleWareHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    if !m.l.GetConn() {
        defer func() { recover() }()
        log.Panicln("Too many requests")
        return
    }
    m.r.ServeHTTP(w, r)
    defer m.l.ReleaseConn()
}

//RegisterHandlers ...
func RegisterHandlers() *httprouter.Router {
    router := httprouter.New()
    router.GET("/ce", ce)
    return router
}

func ce(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
    //为了演示效果这块设置了等待
    time.Sleep(time.Second * 100)
    t, _ := template.ParseFiles("./videos/ce.html")
    t.Execute(w, nil)
}

func main() {
    r := RegisterHandlers()
    //里面的参数2为设置的最大流量
    mh := NewMiddleWareHandler(r, 2)
    http.ListenAndServe(":9000", mh)
}


//ConnLimiter 定义一个结构体
type ConnLimiter struct {
    concurrentConn int
    bucket         chan int
}

//NewConnLimiter ...
func NewConnLimiter(cc int) *ConnLimiter {
    return &ConnLimiter{
        concurrentConn: cc,
        bucket:         make(chan int, cc),
    }
}

//GetConn 获取通道里面的值
func (cl *ConnLimiter) GetConn() bool {
    if len(cl.bucket) >= cl.concurrentConn {
        log.Printf("Reached the rate limitation.")
        return false
    }

    cl.bucket <- 1
    return true
}

//ReleaseConn 释放通道里面的值
func (cl *ConnLimiter) ReleaseConn() {
    c := <-cl.bucket
    log.Printf("New connction coming: %d", c)
}

文献

golang版本实现限速参考
 算法介绍
 tollbooth 一个开箱即用的限速项目
 uber漏铜
 限速