漏桶算法队列未满时实现固定请求处理速率的正确逻辑是什么？

漏桶算法实现逻辑修正建议 #

你现有伪代码主要存在两个核心问题，调整后就能实现正确的流量整形效果：

• 不要在请求处理线程中直接sleep，高并发场景下会快速耗尽服务的线程/协程资源，正确的漏桶实现需要把「请求入队」和「请求执行」两个逻辑完全解耦，避免阻塞请求链路。
• 计算请求执行时间的逻辑有漏洞，不应该取队列首个元素的时间加间隔，应该取队列中最后一个待执行请求的允许执行时间加间隔，才能保证所有请求的执行间隔严格等于你设定的固定值。

队列未满时固定速率整形的正确流程 #

1. 每个请求进来先校验对应客户端标识（IP、用户Token等）的Redis有序集合长度，超过设定的队列容量直接丢弃请求。
2. 队列未满时，先取出当前有序集合中score最大的元素（即最后一个待执行请求的允许执行时间戳），如果集合为空就用当前时间作为计算基准。
3. 计算当前新请求的允许执行时间：如果基准时间大于当前时间，就用基准时间 + 固定间隔作为当前请求的执行时间；如果基准时间小于等于当前时间，说明桶处于空闲状态，直接用当前时间作为执行时间，不需要等待。
4. 把当前请求的唯一标识作为member，允许执行时间作为score写入Redis有序集合，直接返回入队成功响应即可，不要阻塞当前请求线程。如果是同步处理场景，可以让客户端通过请求ID轮询执行结果，或者设置最大等待时间短时间阻塞。
5. 单独启动一个常驻后台协程，用定时器固定每隔你设定的间隔扫一次所有客户端的有序集合，取出所有score小于等于当前时间的请求，按score从小到大依次执行，执行完成后把对应元素从集合中删除即可。

简化版Golang+Redis核心实现代码 #

import (
    "context"
    "net/http"
    "strconv"
    "time"
    "github.com/go-redis/redis/v8"
    "github.com/google/uuid"
)

// 全局常量定义
const (
    BucketCapacity    = 5               // 漏桶队列最大容量
    ProcessInterval   = 7 * time.Second // 固定请求处理间隔
    RedisKeyPrefix    = "leaky_bucket:"
)

var redisClient *redis.Client

// 请求入队逻辑
func enqueueRequest(clientID string, req *http.Request) (bool, error) {
    key := RedisKeyPrefix + clientID
    ctx := context.Background()

    // 检查当前队列长度
    size, err := redisClient.ZCard(ctx, key).Result()
    if err != nil {
        return false, err
    }
    if size >= BucketCapacity {
        return false, nil // 队列满直接丢弃请求
    }

    // 取队列中最后一个待执行请求的执行时间
    lastExecTime := time.Now().Unix()
    lastItems, err := redisClient.ZRevRangeWithScores(ctx, key, 0, 0).Result()
    if err == nil && len(lastItems) > 0 {
        lastExecTime = int64(lastItems[0].Score)
    }

    // 计算当前请求的实际执行时间
    currentExecTime := time.Now().Unix()
    if lastExecTime > currentExecTime {
        currentExecTime = lastExecTime + int64(ProcessInterval.Seconds())
    }

    // 写入Redis有序集合
    reqID := uuid.New().String()
    _, err = redisClient.ZAdd(ctx, key, &redis.Z{
        Score:  float64(currentExecTime),
        Member: reqID,
    }).Result()
    if err != nil {
        return false, err
    }
    // 此处可将reqID与请求的对应关系存入本地或Redis，供后续执行时调用
    return true, nil
}

// 队列消费协程，独立于请求链路运行
func processQueueWorker() {
    ticker := time.NewTicker(ProcessInterval)
    defer ticker.Stop()
    for range ticker.C {
        ctx := context.Background()
        now := time.Now().Unix()
        // 生产环境建议用SCAN命令遍历Key，避免阻塞Redis
        keys, err := redisClient.Keys(ctx, RedisKeyPrefix+"*").Result()
        if err != nil {
            continue
        }
        for _, key := range keys {
            // 取出所有到执行时间的请求
            items, err := redisClient.ZRangeByScoreWithScores(ctx, key, &redis.ZRangeBy{
                Min: "-inf",
                Max: strconv.FormatInt(now, 10),
            }).Result()
            if err != nil || len(items) == 0 {
                continue
            }
            for _, item := range items {
                reqID := item.Member.(string)
                // 取出请求对象执行业务逻辑 processRequest(reqID)
                // 执行完成后删除队列中的对应元素
                redisClient.ZRem(ctx, key, reqID)
            }
        }
    }
}

额外注意事项 #

• 入队的Redis操作建议用Lua脚本打包执行，保证原子性，避免并发请求入队时出现执行时间计算的竞态问题。
• 如果业务允许异步处理，优先用异步解耦的架构，不要在请求链路中加阻塞逻辑，能大幅提升服务吞吐量。
• 如果必须同步返回请求结果，要给阻塞等待设置最大超时时间，避免连接泄漏。

内容的提问来源于stack exchange，提问作者Jason