Mastering Rate Limiting: Leaky Bucket, Token Bucket, and Sliding Window in Go

1. Leaky Bucket Algorithm

Algorithm Idea

The leaky bucket algorithm is the opposite of the token bucket; incoming requests are placed into a bucket, and workers remove requests at a fixed rate for processing. If the bucket is full, the request is rejected with a rate‑limit error.

Applicable Scenarios

It provides the most uniform traffic shaping and is typically used for flow shaping, such as protecting a database. Requests are queued in the bucket and workers consume them at the database’s sustainable QPS. It is not suitable for flash‑sale or hot‑topic scenarios because it handles bursts poorly and requires a separate queue per user/IP, leading to high resource consumption.

Go Implementation

package main

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

type Task struct {
    handler func() Result
    resChan chan Result
    taskID  int
}

type Result struct {}

func handler() Result {
    time.Sleep(300 * time.Millisecond)
    return Result{}
}

func NewTask(id int) Task {
    return Task{handler: handler, resChan: make(chan Result), taskID: id}
}

type LeakyBucket struct {
    BucketSize int
    NumWorker  int
    bucket     chan Task
}

func NewLeakyBucket(bucketSize int, numWorker int) *LeakyBucket {
    return &LeakyBucket{BucketSize: bucketSize, NumWorker: numWorker, bucket: make(chan Task, bucketSize)}
}

func (b *LeakyBucket) validate(task Task) bool {
    select {
    case b.bucket <- task:
    default:
        fmt.Printf("request[id=%d] is refused
", task.taskID)
        return false
    }
    <-task.resChan
    fmt.Printf("request[id=%d] is run
", task.taskID)
    return true
}

func (b *LeakyBucket) Start() {
    go func() {
        for i := 0; i < b.NumWorker; i++ {
            go func() {
                for {
                    task := <-b.bucket
                    result := task.handler()
                    task.resChan <- result
                }
            }()
        }
    }()
}

func main() {
    bucket := NewLeakyBucket(10, 4)
    bucket.Start()

    var wg sync.WaitGroup
    for i := 0; i < 20; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            task := NewTask(id)
            bucket.validate(task)
        }(i)
    }
    wg.Wait()
}

2. Token Bucket Algorithm

Algorithm Idea

Imagine a bucket that receives tokens at a fixed rate; once full, no more tokens are added. When a request arrives, the service tries to take a token; if successful, it proceeds, otherwise it returns a rate‑limit error and stops processing.

Features

Because a request can be processed as long as a token exists, the token bucket handles burst traffic. The fixed token‑adding rate and bucket capacity limit the number of requests per unit time. For example, with a rate of 1 token per 10 ms and a capacity of 500, the bucket can accumulate up to 500 tokens during low traffic; a burst can then consume all tokens, after which new requests are allowed only at the token‑generation rate.

Parameter Settings

Bucket size – should reflect the resource consumption of the business logic and the machine’s concurrency capacity. Token generation rate – if too slow, it cannot accumulate enough tokens to handle bursts.

Applicable Scenarios

Suitable for flash‑sale or hot‑topic scenarios because it can limit traffic while still handling bursts. Uniform‑rate algorithms would block many users during spikes, harming user experience.

Go Implementation

package main

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

var recordMu map[string]*sync.RWMutex

func init() { recordMu = make(map[string]*sync.RWMutex) }

func max(a, b int) int { if a > b { return a }; return b }

type TokenBucket struct {
    BucketSize int
    TokenRate  time.Duration
    records    map[string]*record
}

type record struct { last time.Time; token int }

func NewTokenBucket(bucketSize int, tokenRate time.Duration) *TokenBucket {
    return &TokenBucket{BucketSize: bucketSize, TokenRate: tokenRate, records: make(map[string]*record)}
}

func (t *TokenBucket) getUidOrIp() string { return "127.0.0.1" }

func (t *TokenBucket) getRecord(uidOrIp string) *record {
    if r, ok := t.records[uidOrIp]; ok { return r }
    return &record{}
}

func (t *TokenBucket) storeRecord(uidOrIp string, r *record) { t.records[uidOrIp] = r }

func (t *TokenBucket) validate(uidOrIp string) bool {
    rl, ok := recordMu[uidOrIp]
    if !ok { var mu sync.RWMutex; rl = μ recordMu[uidOrIp] = rl }
    rl.Lock(); defer rl.Unlock()

    r := t.getRecord(uidOrIp)
    now := time.Now()
    if r.last.IsZero() { r.last, r.token = now, t.BucketSize } else if r.last.Add(t.TokenRate).Before(now) {
        r.token += max(int(now.Sub(r.last)/t.TokenRate), t.BucketSize)
        r.last = now
    }
    var result bool
    if r.token > 0 { r.token--; result = true }
    t.storeRecord(uidOrIp, r)
    return result
}

func (t *TokenBucket) IsLimited() bool { return !t.validate(t.getUidOrIp()) }

func main() {
    tokenBucket := NewTokenBucket(5, 100*time.Millisecond)
    for i := 0; i < 6; i++ { fmt.Println(tokenBucket.IsLimited()) }
    time.Sleep(100 * time.Millisecond)
    fmt.Println(tokenBucket.IsLimited())
}

3. Sliding Window Algorithm

Algorithm Idea

The sliding window algorithm improves on a fixed time‑window counter. In a fixed window, each user/IP stores a timestamp and request count. When the window expires, the count resets, which can miss bursts that span the boundary. Sliding windows divide the interval into smaller slots to accurately track bursts.

Applicable Scenarios

Like the token bucket, it can handle burst traffic while providing rate limiting.

Go Implementation

package main

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

var winMu map[string]*sync.RWMutex

func init() { winMu = make(map[string]*sync.RWMutex) }

type timeSlot struct { timestamp time.Time; count int }

func countReq(win []*timeSlot) int { var c int; for _, ts := range win { c += ts.count }; return c }

type SlidingWindowLimiter struct {
    SlotDuration time.Duration
    WinDuration  time.Duration
    numSlots     int
    windows      map[string][]*timeSlot
    maxReq       int
}

func NewSliding(slotDuration, winDuration time.Duration, maxReq int) *SlidingWindowLimiter {
    return &SlidingWindowLimiter{SlotDuration: slotDuration, WinDuration: winDuration, numSlots: int(winDuration/slotDuration), windows: make(map[string][]*timeSlot), maxReq: maxReq}
}

func (l *SlidingWindowLimiter) getWindow(uidOrIp string) []*timeSlot {
    if win, ok := l.windows[uidOrIp]; ok { return win }
    return make([]*timeSlot, 0, l.numSlots)
}

func (l *SlidingWindowLimiter) storeWindow(uidOrIp string, win []*timeSlot) { l.windows[uidOrIp] = win }

func (l *SlidingWindowLimiter) validate(uidOrIp string) bool {
    mu, ok := winMu[uidOrIp]
    if !ok { var m sync.RWMutex; mu = &m; winMu[uidOrIp] = mu }
    mu.Lock(); defer mu.Unlock()

    win := l.getWindow(uidOrIp)
    now := time.Now()
    timeoutOffset := -1
    for i, ts := range win {
        if ts.timestamp.Add(l.WinDuration).After(now) { break }
        timeoutOffset = i
    }
    if timeoutOffset > -1 { win = win[timeoutOffset+1:] }

    var result bool
    if countReq(win) < l.maxReq { result = true }

    var lastSlot *timeSlot
    if len(win) > 0 {
        lastSlot = win[len(win)-1]
        if lastSlot.timestamp.Add(l.SlotDuration).Before(now) {
            lastSlot = &timeSlot{timestamp: now, count: 1}
            win = append(win, lastSlot)
        } else {
            lastSlot.count++
        }
    } else {
        lastSlot = &timeSlot{timestamp: now, count: 1}
        win = append(win, lastSlot)
    }
    l.storeWindow(uidOrIp, win)
    return result
}

func (l *SlidingWindowLimiter) getUidOrIp() string { return "127.0.0.1" }

func (l *SlidingWindowLimiter) IsLimited() bool { return !l.validate(l.getUidOrIp()) }

func main() {
    limiter := NewSliding(100*time.Millisecond, time.Second, 10)
    for i := 0; i < 5; i++ { fmt.Println(limiter.IsLimited()) }
    time.Sleep(100 * time.Millisecond)
    for i := 0; i < 5; i++ { fmt.Println(limiter.IsLimited()) }
    fmt.Println(limiter.IsLimited())
    for _, v := range limiter.windows[limiter.getUidOrIp()] { fmt.Println(v.timestamp, v.count) }
    fmt.Println("a thousand years later...")
    time.Sleep(time.Second)
    for i := 0; i < 7; i++ { fmt.Println(limiter.IsLimited()) }
    for _, v := range limiter.windows[limiter.getUidOrIp()] { fmt.Println(v.timestamp, v.count) }
}