如何在Golang中实现一个基于令牌桶算法的限流器

sync.Mutex

time.Ticker

time.NewTimer

package main

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

// TokenBucket represents a token bucket rate limiter.
type TokenBucket struct {
    capacity    int66 // The maximum number of tokens the bucket can hold.
    rate        int66 // The rate at which tokens are added to the bucket (tokens per second).
    tokens      int66 // The current number of tokens in the bucket.
    lastRefill  time.Time // The last time the bucket was refilled.
    mu          sync.Mutex // Mutex to protect access to the bucket's state.
}

// NewTokenBucket creates and initializes a new TokenBucket.
// capacity: maximum tokens.
// rate: tokens per second.
func NewTokenBucket(capacity, rate int66) *TokenBucket {
    if capacity <= 0 || rate <= 0 {
        panic("capacity and rate must be positive")
    }
    return &TokenBucket{
        capacity:   capacity,
        rate:       rate,
        tokens:     capacity, // Start with a full bucket.
        lastRefill: time.Now(),
    }
}

// Allow checks if a request can be processed.
// It attempts to consume one token. Returns true if successful, false otherwise.
func (tb *TokenBucket) Allow() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    // Calculate how many tokens should have been added since last refill.
    now := time.Now()
    // Using float64 for precision in duration calculation
    duration := now.Sub(tb.lastRefill).Seconds()
    tokensToAdd := int66(duration * float64(tb.rate))

    if tokensToAdd > 0 {
        tb.tokens += tokensToAdd
        if tb.tokens > tb.capacity {
            tb.tokens = tb.capacity // Cap tokens at capacity.
        }
        tb.lastRefill = now
    }

    if tb.tokens >= 1 {
        tb.tokens--
        return true
    }
    return false
}

func main() {
    // Create a token bucket with capacity 10 and refill rate 2 tokens/second.
    limiter := NewTokenBucket(10, 2)

    fmt.Println("Testing Token Bucket Limiter:")
    for i := 0; i < 20; i++ {
        time.Sleep(200 * time.Millisecond) // Simulate requests coming in
        if limiter.Allow() {
            fmt.Printf("Request %d: ALLOWED. Current tokens: %d\n", i+1, limiter.tokens)
        } else {
            fmt.Printf("Request %d: DENIED. Current tokens: %d\n", i+1, limiter.tokens)
        }
    }

    fmt.Println("\nTesting burst capability:")
    limiter = NewTokenBucket(5, 1) // Reset limiter for burst test
    fmt.Println("First 5 requests (burst):")
    for i := 0; i < 5; i++ {
        if limiter.Allow() {
            fmt.Printf("Request %d: ALLOWED. Current tokens: %d\n", i+1, limiter.tokens)
        } else {
            fmt.Printf("Request %d: DENIED. Current tokens: %d\n", i+1, limiter.tokens)
        }
    }
    fmt.Println("Next requests after a short delay:")
    time.Sleep(2 * time.Second) // Wait for tokens to refill
    for i := 0; i < 3; i++ {
        if limiter.Allow() {
            fmt.Printf("Request %d: ALLOWED. Current tokens: %d\n", i+6, limiter.tokens)
        } else {
            fmt.Printf("Request %d: DENIED. Current tokens: %d\n", i+6, limiter.tokens)
        }
    }
}

Allow()

tokens

capacity

tokens

true

INCR

EXPIRE

capacity

rate

prometheus

Allow()

true

allowed_total

false

denied_total

denied_total