在泄漏桶算法中，当队列未满时，实现固定速率的正确逻辑是什么？

Question

我正在学习泄漏的桶算法，并希望通过编写一些带有redis和golang http的简单代码来弄脏我的手。当我在这里搜索关键字redis，泄漏，桶。[1]中有许多类似的问题，这很好。然而，我发现在浏览了这些线程和wiki之后，我有一个问题来理解整个逻辑[2]。我想有些东西我不明白，也没有意识到这一点。因此，我想在这里再次改写它。如果我弄错了，请纠正我。伪代码：key := "ip address, token or anything that can be the representative of a client"redis_queue_size := 5interval_between_each_request := 7request := obtain_http_request_from_somewhere()if check_current_queue_size() < redis_queue_size:    if is_queue_empty()        add_request_to_the_queue() // zadd "ip1" now() now() // now() is something like seconds, milliseconds or nanoseconds e.g. t = 1        process_request(request)    else        now := get_current_time()        // add_request_to_... retrieves the first element in the queue        // compute the expected timestamp to execute the request and its current time        // e.g. zadd "ip1" <time of the first elment in the queue + interval_between_each_request> now        add_request_to_redis_queue_with_timestamp(now, interval_between_each_request) // e.g. zadd "ip" <timestamp as score> <timestamp a request is allowed to be executed>        // Below function check_the_time_left...() will check how many time left at which the current request need to wait.         // For instance, the first request stored in the queue with the command        //    zadd "ip1" 1 1  // t = 1        // and the second request arrives at t = 4 but it is allowed t be executed at t = 8        //    zadd "ip1" 8 4 // where 4 := now, 8 := 1 + interval_between_each_request        // so the N will be 4         N := check_the_time_left_for_the_current_request_to_execute(now, interval_between_each_request)         sleep(N) // now the request wait for 4 seconds before processing the request        process_request(http_request_obj)else    return // discard request我了解队列已满的部分，然后以下请求将被丢弃。但是，我想我可能会误解当队列未满时，如何重塑传入的请求，以便它可以以固定的速率执行。

Answer 1

1. 如果这是为了简单的速率限制，那么使用排序集的滑动窗口方法是我们看到的大多数Redis用户实现的，https://github.com/Redislabs-Solution-Architects/RateLimitingExample/blob/sliding_window/app.py

2. 如果您设置在泄漏的存储桶上，则可以考虑使用每个消费者ID（api令牌/ IP地址等）的redis流，如下所示

请求进入消费者 ID

XADD 请求-[消费者 ID] MAXLEN [存储桶大小]

生成一个 go 例程（如果该使用者 ID 需要），如果请求的 XLEN （[使用者 ID] 为 0，则获取当前时间

XREAD 计数 [number_of_requests_per_period] 块 [时间段 - 1 毫秒] 流请求 -[使用者 ID] 获取当前时间并在剩余时间段内休眠

https://redis.io/commands#stream 详细介绍了流的工作原理

Answer 2

有几种方法可以实现泄漏的存储桶，但该过程应该有两个单独的部分。一个将内容放入存储桶中，另一个在有要删除的内容时按设定的时间间隔删除它们。

您可以使用单独的 goroutine，该 goroutine 将按设定的时间间隔使用消息。这将简化您的代码，因为在一个代码路径上，您只需要查看队列大小并丢弃数据包，而另一个代码路径将仅使用任何存在的内容。