扫码关注官方微信

扫码下载APP
返回顶部
首页 > 资讯 > 后端开发 > GO >go实现一个分布式限流器的方法步骤
  • 831

0
分享到

go实现一个分布式限流器的方法步骤

2024-04-02 19:04:59 831人浏览 独家记忆
摘要

目录1. 接口定义2. LocalCounterLimiter3. LocalTokenBucketLimiter4. RedisCounterLimiter5. RedisToke

项目中需要对 api 的接口进行限流，但是麻烦的是，api 可能有多个节点，传统的本地限流无法处理这个问题。限流的算法有很多，比如计数器法，漏斗法，令牌桶法，等等。各有利弊，相关博文网上很多，这里不再赘述。

项目的要求主要有以下几点：

  • 支持本地/分布式限流，接口统一
  • 支持多种限流算法的切换
  • 方便配置，配置方式不确定

Go 语言不是很支持 OOP，我在实现的时候是按 Java 的思路走的，所以看起来有点不伦不类，希望能抛砖引玉。

1. 接口定义

package ratelimit

import "time"

// 限流器接口
type Limiter interface {
    Acquire() error
    TryAcquire() bool
}

// 限流定义接口
type Limit interface {
    Name() string
    Key() string
    Period() time.Duration
    Count() int32
    LimitType() LimitType
}

// 支持 burst
type BurstLimit interface {
    Limit
    BurstCount() int32
}

// 分布式定义的 burst
type DistLimit interface {
    Limit
    ClusterNum() int32
}

type LimitType int32
const (
    CUSTOM LimitType = iota
    IP
)

Limiter 接口参考了 Google 的 guava 包里的 Limiter 实现。Acquire 接口是阻塞接口，其实还需要加上 context 来保证调用链安全，因为实际项目中并没有用到 Acquire 接口，所以没有实现完善；同理，超时时间的支持也可以通过添加新接口继承自 Limiter 接口来实现。TryAcquire 会立即返回。

Limit 抽象了一个限流定义，Key() 方法返回这个 Limit 的唯一标识，Name() 仅作辅助，Period() 表示周期，单位是秒，Count() 表示周期内的最大次数，LimitType(）表示根据什么来做区分，如 IP，默认是 CUSTOM.

BurstLimit 提供突发的能力，一般是配合令牌桶算法。DistLimit 新增 ClusterNum() 方法，因为 mentor 要求分布式遇到错误的时候，需要退化为单机版本，退化的策略即是：2 节点总共 100QPS，如果出现分区，每个节点需要调整为各 50QPS

2. LocalCounterLimiter

package ratelimit

import (
    "errors"
    "fmt"
    "math"
    "sync"
    "sync/atomic"
    "time"
)

// todo timer 需要 stop
type localCounterLimiter struct {
    limit Limit

    limitCount int32 // 内部使用，对 limit.count 做了 <0 时的转换

    ticker *time.Ticker
    quit chan bool

    lock sync.Mutex
    newTerm *sync.Cond
    count int32
}

func (lim *localCounterLimiter) init() {
    lim.newTerm = sync.NewCond(&lim.lock)
    lim.limitCount = lim.limit.Count()

    if lim.limitCount < 0 {
        lim.limitCount = math.MaxInt32 // count 永远不会大于 limitCount，后面的写法保证溢出也没问题
    } else if lim.limitCount == 0  {
        // 禁止访问, 会无限阻塞
    } else {
        lim.ticker = time.NewTicker(lim.limit.Period())
        lim.quit = make(chan bool, 1)

        go func() {
            for {
                select {
                case <- lim.ticker.C:
                    fmt.Println("ticker .")
                    atomic.StoreInt32(&lim.count, 0)
                    lim.newTerm.Broadcast()

                    //lim.newTerm.L.Unlock()
                case <- lim.quit:
                    fmt.Println("work well .")
                    lim.ticker.Stop()
                    return
                }
            }
        }()
    }
}

// todo 需要机制来防止无限阻塞, 不超时也应该有个极限时间
func (lim *localCounterLimiter) Acquire() error {
    if lim.limitCount == 0 {
        return errors.New("rate limit is 0, infinity wait")
    }

    lim.newTerm.L.Lock()
    for lim.count >= lim.limitCount {
        // block instead of spinning
        lim.newTerm.Wait()
        //fmt.Println(count, lim.limitCount)
    }
    lim.count++
    lim.newTerm.L.Unlock()

    return nil
}

func (lim *localCounterLimiter) TryAcquire() bool {
    count := atomic.AddInt32(&lim.count, 1)
    if count > lim.limitCount {
        return false
    } else {
        return true
    }
}

代码很简单，就不多说了

3. LocalTokenBucketLimiter

golang 的官方库里提供了一个 ratelimiter，就是采用令牌桶的算法。所以这里并没有重复造轮子，直接代理了 ratelimiter。

package ratelimit

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

type localTokenBucketLimiter struct {
    limit Limit

    limiter *rate.Limiter // 直接复用令牌桶的
}

func (lim *localTokenBucketLimiter) init() {
    burstCount := lim.limit.Count()
    if burstLimit, ok := lim.limit.(BurstLimit); ok {
        burstCount = burstLimit.BurstCount()
    }

    count := lim.limit.Count()
    if count < 0 {
        count = math.MaxInt32
    }

    f := float64(count) / lim.limit.Period().Seconds()
    if f < 0 {
        f = float64(rate.Inf) // 无限
    } else if f == 0 {
        panic("为 0 的时候，底层实现有问题")
    }

    lim.limiter = rate.NewLimiter(rate.Limit(f), int(burstCount))
}

func (lim *localTokenBucketLimiter) Acquire() error {
    err := lim.limiter.Wait(context.TODO())
    return err
}

func (lim *localTokenBucketLimiter) TryAcquire() bool {
    return lim.limiter.Allow()
}

4. RedisCounterLimiter

package ratelimit

import (
    "math"
    "sync"
    "xg-go/log"
    "xg-go/xg/common"
)

type redisCounterLimiter struct {
    limit      DistLimit
    limitCount int32 // 内部使用，对 limit.count 做了 <0 时的转换

    redisClient *common.RedisClient

    once sync.Once // 退化为本地计数器的时候使用
    localLim Limiter

    //script string
}

func (lim *redisCounterLimiter) init() {
    lim.limitCount = lim.limit.Count()
    if lim.limitCount < 0 {
        lim.limitCount = math.MaxInt32
    }

    //lim.script = buildScript()
}

//func buildScript() string {
//  sb := strings.Builder{}
//
//  sb.WriteString("local c")
//  sb.WriteString("\nc = redis.call('get',KEYS[1])")
//  // 调用不超过最大值，则直接返回
//  sb.WriteString("\nif c and tonumber(c) > tonumber(ARGV[1]) then")
//  sb.WriteString("\nreturn c;")
//  sb.WriteString("\nend")
//  // 执行计算器自加
//  sb.WriteString("\nc = redis.call('incr',KEYS[1])")
//  sb.WriteString("\nif tonumber(c) == 1 then")
//  sb.WriteString("\nredis.call('expire',KEYS[1],ARGV[2])")
//  sb.WriteString("\nend")
//  sb.WriteString("\nif tonumber(c) == 1 then")
//  sb.WriteString("\nreturn c;")
//
//  return sb.String()
//}

func (lim *redisCounterLimiter) Acquire() error {
    panic("implement me")
}

func (lim *redisCounterLimiter) TryAcquire() (success bool) {
    defer func() {
        // 一般是 redis 连接断了，会触发空指针
        if err := recover(); err != nil {
            //log.Errorw("TryAcquire err", common.ERR, err)
            //success = lim.degradeTryAcquire()
            //return
            success = true
        }

        // 没有错误，判断是否开启了 local 如果开启了，把它停掉
        //if lim.localLim != nil {
        //  // stop 线程安全
        //  lim.localLim.Stop()
        //}
    }()

    count, err := lim.redisClient.IncrBy(lim.limit.Key(), 1)
    //panic("模拟 redis 出错")
    if err != nil {
        log.Errorw("TryAcquire err", common.ERR, err)
        panic(err)
    }

    // *2 是为了保留久一点，便于观察
    err = lim.redisClient.Expire(lim.limit.Key(), int(2 * lim.limit.Period().Seconds()))
    if err != nil {
        log.Errorw("TryAcquire error", common.ERR, err)
        panic(err)
    }

    // 业务正确的情况下 确认超限
    if int32(count) > lim.limitCount {
        return false
    }

    return true

    //keys := []string{lim.limit.Key()}
    //
    //log.Errorw("TryAcquire ", keys, lim.limit.Count(), lim.limit.Period().Seconds())
    //count, err := lim.redisClient.Eval(lim.script, keys, lim.limit.Count(), lim.limit.Period().Seconds())
    //if err != nil {
    //  log.Errorw("TryAcquire error", common.ERR, err)
    //  return false
    //}
    //
    //
    //typeName := reflect.TypeOf(count).Name()
    //log.Errorw(typeName)
    //
    //if count != nil && count.(int32) <= lim.limitCount {
    //
    //  return true
    //}
    //return false
}

func (lim *redisCounterLimiter) Stop() {
    // 判断是否开启了 local 如果开启了，把它停掉
    if lim.localLim != nil {
        // stop 线程安全
        lim.localLim.Stop()
    }
}

func (lim *redisCounterLimiter) degradeTryAcquire() bool {
    lim.once.Do(func() {
        count := lim.limit.Count() / lim.limit.ClusterNum()
        limit := LocalLimit {
            name: lim.limit.Name(),
            key: lim.limit.Key(),
            count: count,
            period: lim.limit.Period(),
            limitType: lim.limit.LimitType(),
        }

        lim.localLim = NewLimiter(&limit)
    })

    return lim.localLim.TryAcquire()
}

代码里回退的部分注释了，因为线上为了稳定，实习生的代码毕竟，所以先不跑。

本来原有的思路是直接用 lua 脚本在 redis 上保证原子操作，但是底层封装的库对于直接调 eval 跑的时候，会抛错，而且 source 是 go-redis 里面，赶 ddl 没有时间去 debug，所以只能用 incrBy + expire 分开来。

5. RedisTokenBucketLimiter

令牌桶的状态变量得放在一个 线程安全/一致 的地方，redis 是不二人选。但是令牌桶的算法核心是个延迟计算得到令牌数量，这个是一个很长的临界区，所以要么用分布式锁，要么直接利用 redis 的单线程以原子方式跑。一般业界是后者，即 lua 脚本维护令牌桶的状态变量、计算令牌。代码类似这种

local tokens_key = KEYS[1]
local timestamp_key = KEYS[2]
--redis.log(redis.LOG_WARNING, "tokens_key " .. tokens_key)

local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])
local intval = tonumber(ARGV[5])

local fill_time = capacity/rate
local ttl = math.floor(fill_time*2) * intval

local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
  last_tokens = capacity
end

local last_refreshed = tonumber(redis.call("get", timestamp_key))
if last_refreshed == nil then
  last_refreshed = 0
end

local delta = math.max(0, now-last_refreshed)
local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
local allowed = filled_tokens >= requested
local new_tokens = filled_tokens
if allowed then
  new_tokens = filled_tokens - requested
end

redis.call("setex", tokens_key, ttl, new_tokens)
redis.call("setex", timestamp_key, ttl, now)

return { allowed, new_tokens }

到此这篇关于go实现一个分布式限流器的方法步骤的文章就介绍到这了,更多相关go 分布式限流器内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

--结束END--

本文标题: go实现一个分布式限流器的方法步骤

本文链接: https://lsjlt.com/news/162362.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

猜你喜欢
  • go实现一个分布式限流器的方法步骤
    目录1. 接口定义2. LocalCounterLimiter3. LocalTokenBucketLimiter4. RedisCounterLimiter5. RedisToke...
    99+
    2024-04-02
  • Go实现分布式系统高可用限流器实战
    目录前言1. 问题描述2. 信号量限流2.1 阻塞方式2.2 非阻塞方式3. 限流算法3.1 漏桶算法3.2 令牌桶算法3.3 漏桶算法的实现改进4. Uber 开源实现 RateL...
    99+
    2024-04-02
  • 基于Redis实现分布式应用限流的方法
    限流的目的是通过对并发访问/请求进行限速或者一个时间窗口内的的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务。前几天在DD的公众号，看了一篇关于使用 瓜娃 实现单应用限流的方案 --》原文，参考《redis in action》 实...
    99+
    2023-05-30
    redis 限流 流的
  • Redis和Lua实现分布式限流器
    本篇文章为大家展示了Redis和Lua实现分布式限流器，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。原理计数器算法是指在一段窗口时间内允许通过的固定数量的请求, 比...
    99+
    2024-04-02
  • SpringBoot+Redis+Lua分布式限流的实现
    Redis支持LUA脚本的主要优势 LUA脚本的融合将使Redis数据库产生更多的使用场景，迸发更多新的优势： 高效性：减少网络开销及时延，多次redis服务器网络请求的操作，使用L...
    99+
    2022-11-13
    SpringBoot Redis Lua分布式限流 SpringBoot 分布式限流
  • 四种分布式限流算法和代码实现
    带着问题走近限流为什么要限流呢？就像我上面说的，流量多，的确是一件好事，但是如果过载，把系统打挂了，那大家都要吃席了。没逝吧所以，在各种大促活动之前，要对系统进行压测，评估整个系统的峰值QPS，要做一些限流的设置，超过一定阈值，就拒绝处理或...
    99+
    2023-08-15
  • 打包教程：Go语言分布式编程中的一个重要步骤？
    Go语言是一种开源的编程语言，其具有高效、可靠、快速的特性，因此在分布式编程中被广泛应用。在Go语言分布式编程中，打包是一个重要的步骤，本文将为您介绍Go语言分布式编程中打包的相关知识。 一、什么是打包？ 在Go语言中，打包就是将一个或多个...
    99+
    2023-09-27
    分布式 教程 打包
  • Go实现各类限流的方法
    前 言 在开发高并发系统时，我们可能会遇到接口访问频次过高，为了保证系统的高可用和稳定性，这时候就需要做流量限制，你可能是用的 Nginx 这种来控制请求，...
    99+
    2022-06-07
    GO 方法
  • Redisson分布式限流的实现原理解析
    目录正文RRateLimiter使用RRateLimiter的实现RRateLimiter使用时注意事项RRateLimiter是非公平限流器Rate不要设置太大限流的上限取决于Redis单实例的性能分布式限流的本质正文...
    99+
    2023-02-12
    Redisson分布式限流 Redisson 分布式
  • Go语言开发实现分布式流式计算系统的方法与实践
    Go语言是一种自由、开源的编程语言，它以其高效的并发模型和简洁的代码风格而广受开发者的喜爱。在分布式计算领域，Go语言也展现出了其强大的开发能力和适用性。本文将介绍使用Go语言开发实现分布式流式计算系统的方法与实践。一、分布式流式计算系统概...
    99+
    2023-11-20
    分布式 方法和实践 流式计算
  • SpringCloud实现灰度发布的方法步骤
    目录1.什么是灰度发布？2.灰度发布有什么作用？3.灰度发布的实现方式：网关到服务，服务到服务3.1网关到服务代码实现3.2网关到服务代码实现1.什么是灰度发布？ 灰度发布又称金丝雀...
    99+
    2024-04-02
  • Go语言实战之实现一个简单分布式系统
    目录引子思路实战节点通信主节点工作节点将它们放在一起代码效果总结引子 如今很多云原生系统、分布式系统，例如 Kubernetes，都是用 Go 语言写的，这是因为 Go 语言天然支持...
    99+
    2024-04-02
  • web开发中分布式系统中的限流器实现算法怎么用
    本篇文章给大家分享的是有关web开发中分布式系统中的限流器实现算法怎么用，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。一般限流器有五种算法，分...
    99+
    2024-04-02
  • redis实现分布式锁的方法
    本篇文章展示了redis实现分布式锁的方法具体操作，代码简明扼要容易理解，绝对能让你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。分布式锁其实可以理解为：控制分布式系统有序的去对共享资源进行操作，通过互...
    99+
    2024-04-02
  • 实现一个响应式布局的原理与方法
    页面响应式布局的原理与实现方法 随着移动设备的普及和互联网的快速发展，越来越多的用户开始使用手机、平板等移动设备浏览网页。而传统的固定布局往往无法适应不同屏幕尺寸的设备，导致用户体验不佳。为了解决这个问题，响应式布局应运而生。 ...
    99+
    2024-01-29
    原理 实现方法 响应式布局
  • Go实现分布式唯一ID的生成之雪花算法
    目录背景：特性:雪花算法:分布式唯一ID的生成 背景： 在分布式架构下,唯一序列号生成是我们在设计一个尤其是数据库使用分库分表的时候会常见的一个问题 特性: 全局唯一,这是基本要求,...
    99+
    2024-04-02
  • 阿里云一台跑两个服务器的实现方法与步骤
    在云计算的环境中，一台服务器通常需要支持多个虚拟机或者容器来运行多个应用，以提高资源利用效率。本文将详细说明如何在阿里云上实现一台服务器跑两个服务器的操作步骤，以帮助读者更好地理解并使用云计算。 在阿里云上，一台服务器可以运行多个服务器的实...
    99+
    2023-11-10
    一台 阿里 步骤
  • springboot+redis 实现分布式限流令牌桶的示例代码
    1、前言 网上找了很多redis分布式限流方案，要不就是太大，需要引入第三方jar，而且还无法正常运行，要不就是定时任务定时往key中放入数据，使用的时候调用，严重影响性能，所以着手...
    99+
    2024-04-02
  • Go 实现 Nginx 加权轮询算法的方法步骤
    目录一，Nginx 负载均衡的轮询 （round-robin）1. nginx 中的配置2. 简单介绍3. 特点4. 实现 （这里使用golang模拟实现）5. 测试二，Ngi...
    99+
    2022-06-07
    GO 方法 算法 Nginx
  • redis中分布式锁的实现方法
    小编给大家分享一下redis中分布式锁的实现方法，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！redis分布式锁：1、实现原理利...
    99+
    2024-04-02
软考高级职称资格查询
热门wiki
近期文章
推荐阅读
热门问答
热门标签
编程网，编程工程师的家园，是目前国内优秀的开源技术社区之一，形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容，为IT开发者提供了一个发现、使用、并交流开源技术的平台。

  • 官方手机版

  • 微信公众号

  • 商务合作