如何用Golang实现并发限流演示令牌桶与漏桶算法的Go实现

P粉602998670

发布时间：2025-08-17 12:36:02

855人浏览过

来源于php中文网

原创

令牌桶和漏桶是实现并发限流的两种经典算法。1. 令牌桶通过定时添加令牌、请求获取令牌执行，允许突发流量；2. 漏桶则以固定速率处理请求队列，严格控制流量。两者均可用go语言通过channel和定时器实现，适用于不同限流场景。

如何用Golang实现并发限流演示令牌桶与漏桶算法的Go实现

实现并发限流是高并发系统中常见的需求，Golang作为一门天然支持并发的语言，在这方面有很好的表现。其中，令牌桶和漏桶算法是比较经典的限流算法。下面将通过简单的示例来演示这两种算法的Go语言实现。

什么是令牌桶限流？

令牌桶（Token Bucket）是一种动态限流算法，其核心思想是：系统以固定速率向桶中添加令牌，请求只有在获取到令牌时才能被处理。如果桶满了，多余的令牌会被丢弃；如果没有令牌了，请求就会被拒绝或排队等待。

实现思路：

使用一个带缓冲的channel模拟令牌桶
启动一个goroutine定时往channel中放入令牌
每次请求尝试从channel中取出令牌，取不到则拒绝

package main

import (
    "fmt"
    "sync"
    "time"
)

func tokenBucket(rps int) chan struct{} {
    tokens := make(chan struct{}, rps)

    // 初始填充令牌
    for i := 0; i < rps; i++ {
        tokens <- struct{}{}
    }

    go func() {
        ticker := time.NewTicker(time.Second / time.Duration(rps))
        defer ticker.Stop()

        for range ticker.C {
            select {
            case tokens <- struct{}{}:
            default:
                // 桶满就丢弃
            }
        }
    }()

    return tokens
}

func main() {
    rate := 3 // 每秒允许3个请求
    tokens := tokenBucket(rate)

    var wg sync.WaitGroup

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            time.Sleep(time.Millisecond * 200) // 模拟并发请求
            select {
            case <-tokens:
                fmt.Printf("请求 %d 被处理\n", id)
            default:
                fmt.Printf("请求 %d 被限流\n", id)
            }
        }(i)
    }

    wg.Wait()
}

什么是漏桶限流？

漏桶（Leaky Bucket）算法则是另一种限流方式。它的原理是：所有请求先进入一个队列（桶），然后系统以固定速率从队列中取出请求处理。即使短时间内大量请求涌入，也只能按照固定的速率处理。

Designs.ai

AI设计工具

下载

立即学习“go语言免费学习笔记（深入）”；

实现思路：

维护一个请求队列
定时从队列中取出请求进行处理
队列满了之后拒绝新请求

package main

import (
    "fmt"
    "sync"
    "time"
)

type LeakyBucket struct {
    capacity  int           // 桶容量
    rate      time.Duration // 出水速率
    queue     chan struct{}
    wg        sync.WaitGroup
}

func NewLeakyBucket(capacity int, rate time.Duration) *LeakyBucket {
    lb := &LeakyBucket{
        capacity: capacity,
        rate:     rate,
        queue:    make(chan struct{}, capacity),
    }

    lb.startDraining()
    return lb
}

func (lb *LeakyBucket) startDraining() {
    go func() {
        ticker := time.NewTicker(lb.rate)
        defer ticker.Stop()

        for range ticker.C {
            select {
            case <-lb.queue:
                fmt.Println("处理一个请求")
            default:
            }
        }
    }()
}

func (lb *LeakyBucket) HandleRequest() bool {
    select {
    case lb.queue <- struct{}{}:
        return true
    default:
        return false
    }
}

func main() {
    bucket := NewLeakyBucket(5, time.Second/2) // 容量5，每秒处理2个请求

    for i := 0; i < 10; i++ {
        time.Sleep(200 * time.Millisecond)
        if bucket.HandleRequest() {
            fmt.Printf("请求 %d 进入队列\n", i)
        } else {
            fmt.Printf("请求 %d 被拒绝\n", i)
        }
    }

    time.Sleep(6 * time.Second) // 等待漏桶处理完剩余请求
}