在 golang 分布式系统中，协程可用于实现限流和熔断。限流通过令牌桶算法限制并发访问量，熔断则采用断路器模式，当故障频繁时暂时停止访问。限流和熔断机制可防止系统过载或崩溃，保证系统的稳定和响应能力。

利用 Go 协程实现限流和熔断

在分布式系统中，限制对资源的并发访问量和处理故障非常重要。Golang 中的协程提供了一个轻量级的并行机制，可用于轻松实现限流和熔断。

限流

Golang Uber Dig

使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等

下载

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

限制对资源的并发访问量可以防止系统因过载而崩溃。可以使用令牌桶算法来实现限流：

package main

import (
    "context"
    "fmt"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)

func main() {
    // 令牌生成速率（令牌/秒）
    rate := 100

    // 令牌桶容量
    capacity := 1000

    // 创建令牌桶
    bucket := NewTokenBucket(rate, capacity)

    // 模拟并发请求
    wg := sync.WaitGroup{}
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func(requestID int) {
            defer wg.Done()

            if !bucket.TryAcquire() {
                // 如果无法获取令牌，则丢弃请求
                fmt.Println("Request", requestID, "dropped due to rate limiting")
            } else {
                // 处理请求
                fmt.Println("Request", requestID, "processed")
            }
        }(i)
    }

    wg.Wait()
}

// 令牌桶
type TokenBucket struct {
    rate     int
    capacity int

    tokens      int64
    lastUpdated time.Time
    lock        sync.RWMutex
}

// NewTokenBucket 创建一个新的令牌桶
func NewTokenBucket(rate int, capacity int) *TokenBucket {
    bucket := &TokenBucket{
        rate:     rate,
        capacity: capacity,

        tokens:      capacity,
        lastUpdated: time.Now(),
    }

    // 启动定时任务更新令牌
    go bucket.Tick()

    return bucket
}

// TryAcquire 尝试获取一个令牌
func (b *TokenBucket) TryAcquire() bool {
    for {
        b.lock.Lock()
        tokens := b.tokens

        // 计算自上次更新以来经过的时间
        elapsed := time.Since(b.lastUpdated)
        // 根据时间更新令牌
        newTokens := b.rate * int(elapsed.Seconds())
        // 更新令牌量
        tokens += newTokens
        // 确保令牌量不超过容量
        tokens = min(tokens, b.capacity)
        // 更新最后更新时间
        b.lastUpdated = time.Now()

        if tokens > 0 {
            tokens--
            atomic.StoreInt64(&b.tokens, tokens)
            b.lock.Unlock()
            return true
        }

        b.lock.Unlock()
        return false
    }
}

// Tick 定时任务更新令牌
func (b *TokenBucket) Tick() {
    ticker := time.NewTicker(100 * time.Millisecond)
    for {
        select {
        case <-ticker.C:
            b.TryAcquire()
        }
    }
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

熔断

熔断是指当资源不可用或响应速度过慢时，临时停止对该资源的访问。这可以防止不必要的请求堆积，从而导致系统崩溃。可以使用断路器模式来实现熔断：

package main

import (
    "context"
    "fmt"
    "sync/atomic"
    "time"
)

func main() {
    // 连续失败的请求次数阈值
    failureThreshold := 5

    // 熔断持续时间（秒）
    timeout := 30

    // 创建熔断器
    breaker := NewCircuitBreaker(failureThreshold, timeout)

    // 模拟并发请求
    wg := sync.WaitGroup{}
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func(requestID int) {
            defer wg.Done()

            // 尝试执行请求
            if breaker.Call(func() error {
                // 实际的请求处理
                return nil
            }) {
                // 请求成功
                fmt.Println("Request", requestID, "processed")
            } else {
                // 请求被熔断
                fmt.Println("Request", requestID, "dropped due to circuit breaker")
            }
        }(i)
    }

    wg.Wait()
}

// 熔断器
type CircuitBreaker struct {
    failureThreshold int
    timeout          time.Duration

    state             atomic.Value
    lastFailureAt     atomic.Value
    failureCount      int32
    resetTimerStarted bool
}

// NewCircuitBreaker 创建一个新的熔断器
func NewCircuitBreaker(failureThreshold int, timeout time.Duration) *CircuitBreaker {
    breaker := &CircuitBreaker{
        failureThreshold: failureThreshold,
        timeout:          timeout,
    }

    // 初始化熔断器状态
    breaker.SetState(Closed)

    return breaker
}

// SetState 设置熔断器状态
func (b *CircuitBreaker) SetState(state State) {
    b.state.Store(state)
}

// State 获取熔断器状态
func (b *CircuitBreaker) State() State {
    return b.state.Load()
}

// Call 执行受熔断器保护的函数
func (b *CircuitBreaker) Call(f func() error) error {
    state := b.State()

    switch state {
    case Closed:
        // 熔断器已关闭，尝试执行函数
        return b.execute(f)
    case Open:
        // 熔断器已打开，直接返回错误
        return ErrCircuitOpen
    case HalfOpen:
        // 熔断器处于半开状态，尝试执行函数并更新熔断器状态
        if err := b.execute(f); err != nil {
            b.SetState(Open)
            return err
        } else {
            b.SetState(Closed)
            return nil
        }
    default:
        return ErrUnknownState
    }
}

// execute 执行函数并更新熔断器状态
func (b *CircuitBreaker) execute(f func() error) error {
    // 记录函数调用时间