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

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 := 5

interval_between_each_request := 7

request := 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

如果这是为了简单的速率限制，那么使用排序集的滑动窗口方法是我们看到的大多数Redis用户实现的，https://github.com/Redislabs-Solution-Architects/RateLimitingExample/blob/sliding_window/app.py如果您设置在泄漏的存储桶上，则可以考虑使用每个消费者ID（api令牌/ IP地址等）的redis流，如下所示请求进入消费者 IDXADD 请求-[消费者 ID] MAXLEN [存储桶大小]生成一个 go 例程（如果该使用者 ID 需要），如果请求的 XLEN （[使用者 ID] 为 0，则获取当前时间XREAD 计数 [number_of_requests_per_period] 块 [时间段 - 1 毫秒] 流请求 -[使用者 ID] 获取当前时间并在剩余时间段内休眠https://redis.io/commands#stream&nbsp;详细介绍了流的工作原理

Answer 2

有几种方法可以实现泄漏的存储桶，但该过程应该有两个单独的部分。一个将内容放入存储桶中，另一个在有要删除的内容时按设定的时间间隔删除它们。您可以使用单独的 goroutine，该 goroutine 将按设定的时间间隔使用消息。这将简化您的代码，因为在一个代码路径上，您只需要查看队列大小并丢弃数据包，而另一个代码路径将仅使用任何存在的内容。