速率限制器无法正常工作

Question

我正在编写一个算法来接收消息，然后确定它们是否符合既定的消息传递速率。

例如，在任何 50 秒的窗口中发送的消息不超过 5 条。因此，此窗口必须是滚动窗口。

我已经从this post. 实现了这个令牌桶算法但是，我无法让它始终如一地工作。它通过了一些测试用例，但没有通过其他测试用例，这让我觉得这里隐藏了一个逻辑问题。

这是我目前所拥有的：

public class Messaging {
//times in millis
double time_before;
double time_now;
double now;
double time_passed;

double allowance;

//constants
static double per = 50000; // 50 seconds
static double rate = 5; //5 messages

public Messaging(){
    time_before = System.currentTimeMillis();
    allowance = rate;
}

public void onEvent(){
    time_now = System.currentTimeMillis();
    time_passed = time_now - time_before;
    time_before = time_now;

    allowance += time_passed * (rate / per);

    if (allowance > rate){
        allowance = rate;
        System.out.println("Reset Allowance");
    }

    if (allowance < 1.0){
        System.out.println("Discard");
    }else{
        System.out.println("Forward message");
        allowance -= 1.0;
    }


}

但这不起作用！

public static void main(String[] args) {
    Messaging orders = new Messaging();
    for (int i = 0; i < 10; i++) {
        orders.onEvent();
        try {
            Thread.sleep(5000);
        } catch (Exception ex) {

        }        
    }
}

运行上面的代码会得到：

Forward message. Time: 1469830426910
Forward message. Time: 1469830431912
Forward message. Time: 1469830436913
Forward message. Time: 1469830441920
Forward message. Time: 1469830446929
Forward message. Time: 1469830451937
Forward message. Time: 1469830456939
Forward message. Time: 1469830461952
Forward message. Time: 1469830466962
Discard. Time: 1469830471970
Total time passed: 50067

为什么只丢弃最后一条消息？不应该将津贴减少到足以在第 5 条消息后自动失败吗？

我希望获得有关此特定实施的帮助。实际实现将使用没有队列等的专有语言。

Answer 1

对于速率受限的滑动窗口，您需要将每条消息连同其时间戳一起排队。这样，当队列已满时，您只需丢弃任何新消息。当队列末尾的消息在队列中的停留时间超过了分配的时间时，它们就会离开队列，您就有空间容纳更多新消息。

class MessageBuffer {
    class Message {
        double timestamp;
        String value; // Can be any type you need it to be

        public Message(double timestamp, String value) {
            this.timestamp = timestamp;
            this.value = value;
        }
    }

    static final double WINDOW_SIZE = 5;
    static final double TIME_LIMIT = 50000;

    Queue<Message> messages = new ArrayDeque<>(WINDOW_SIZE);

    public void onEvent(String message) {
        double now = System.currentTimeMillis();

        // If the queue has messages in them that are no longer in the sliding window,
        // remove them from the queue
        while (messages.size() > 0
            && messages.peek().timestamp + TIME_LIMIT > now)
            messages.remove();

        // If there is room in the queue, process this message, otherwise discard it
        if (messages.size() < WINDOW_SIZE) {
            System.out.println("Forward message: " + message);
            messages.add(new Message(now, message));
        } else {
            System.out.println("Discard message: " + message);
        }
    }
}

如果没有这个时间戳信息，您就无法知道消息何时离开滑动窗口，因此您无法知道您的窗口是否已满。仅供参考，您链接到的示例是一个近似值，实际上可以将您限制为 50 秒内少于 5 条消息。

两个小问题：

• 在面向对象的世界中，类是事物，而不是动作。在大多数情况下，你的类名应该是一个描述它是什么的名词（MessageBuffer），而不是一个描述它做什么的动词（Messaging）。
• 变量now不应该是成员变量——当你第一次分配它时，它确实代表当前时间，但是一旦方法完成，并且调用了另一个方法，它的值已过时 - 它实际上并不代表当前时间。

Answer 2

您使用的算法是一个近似值 - 平均而言，它将为您提供您正在寻找的速率。在original post 中，作者声称：

'allowance' 最多以每秒 5/8 个单位的速度增长，即每八秒最多五个单位。转发的每条消息都会扣除一个单位，因此每 8 秒发送的消息不能超过 5 条。

这并不完全正确 - 如果津贴从最大值开始，比如 5，它会以每秒 5/8 个单位增长，然后每发送一条消息，都会减去 1。所以余量以每秒 3/8 个单位的速度减少，从 5 开始，我们可以在限制发生之前发送大约 10 条消息。

如果您在一段时间内收到的消息没有您的节流速度那么快，那么它会增加限额。然后，当消息加快速度时，您有一段短暂的时间，您可能最终会在限制开始之前处理2 * rate 消息。如果您将循环更改为执行 20 次迭代，而不是 10 次，您最终会看到节流的行为与您预期的一样。或者，如果您一开始就将余量设置为 0，而不是速率，它会立即节流。