计算滚动窗口中每秒的消息数？

Question

我的程序中有消息以毫秒分辨率进入（从零到几百条消息/毫秒）。

我想做一些分析。具体来说，我想维护消息计数的多个滚动窗口，并在消息进入时进行更新。例如，

• 最后一秒的消息数、
• 最后一分钟的消息数、
• 最后半小时的消息数除以 过去一小时内的消息数

我不能只维护一个简单的计数，例如 “最后一秒有 1,017 条消息”，因为我不知道消息何时早于 1 秒，因此应该不再在计数中...

我想到维护所有消息的队列，搜索超过一秒的最年轻的消息，并从索引推断计数。然而，这似乎太慢了，并且会消耗大量内存。

我可以做什么来跟踪程序中的这些计数，以便我可以有效地实时获取这些值？

Answer 1

这最容易通过循环缓冲区来处理。

循环缓冲区具有固定数量的元素和指向它的指针。您可以将一个元素添加到缓冲区，当您这样做时，您将增加指向下一个元素的指针。如果超过了固定长度缓冲区，则从头开始。这是存储“最后 N”个项目的一种节省空间和时间的方式。

现在，在您的情况下，您可以拥有一个包含 1,000 个计数器的循环缓冲区，每个计数器计算一毫秒内的消息数量。将所有 1,000 个计数器相加即可得出最后一秒的总计数。当然，您可以通过增量更新计数来优化报告部分，即从计数中扣除插入时覆盖的数字，然后添加新数字。

然后，您可以拥有另一个具有 60 个槽的循环缓冲区，并计算整秒内的消息总数；每秒一次，获取毫秒缓冲区的总计数并将计数写入具有秒分辨率的缓冲区等。

这里是类似 C 的伪代码：

int msecbuf[1000]; // initialized with zeroes
int secbuf[60]; // ditto
int msecptr = 0, secptr = 0;
int count = 0;
int msec_total_ctr = 0;
void msg_received() { count++; }
void every_msec() {
  msec_total_ctr -= msecbuf[msecptr];
  msecbuf[msecptr] = count;
  msec_total_ctr += msecbuf[msecptr];
  count = 0;
  msecptr = (msecptr + 1) % 1000;
}
void every_sec() {
  secbuf[secptr] = msec_total_ctr;
  secptr = (secptr + 1) % 60;
}

Answer 2

您需要指数平滑，也称为指数加权移动平均线。获取自上一条消息到达以来的时间的 EWMA，然后将该时间划分为一秒。您可以使用不同的重量运行其中几个，以有效覆盖更长的时间间隔。实际上，您使用的是无限长的窗口，因此您不必担心数据过期；减肥可以为你做到这一点。

Answer 3

对于最后一个毫秒，保留计数。当毫秒切片转到下一个时，重置计数并将计数添加到毫秒滚动缓冲区数组。如果保持这个累积值，则可以使用固定的内存量提取每秒的消息数。

当 0.1 秒的切片（或紧邻 1 分钟的其他一些小值）完成时，将滚动缓冲区数组中的最后 0.1*1000 个项目相加，并将其放入下一个滚动缓冲区中。通过这种方式，您可以保持较小的毫秒滚动缓冲区（最多 1 秒查找 1000 个项目）和每分钟查找缓冲区（600 个项目）。

您可以以 0.1 分钟为间隔进行整分钟的下一个技巧。所有提出的问题都可以通过对几个整数求和（或者使用 cummulative 时，减去两个值）来查询。

唯一的缺点是最后的秒值每毫秒改变一次，分钟值仅每 0.1 秒改变一次，小时值（以及最后 1/2 小时的 % 的导数）每 0.1 分钟改变一次。但至少你可以控制内存的使用。

Answer 4

您的滚动显示窗口只能更新这么快，假设您想每秒更新 10 次，因此对于 1 秒的数据，您将需要 10 个值。每个值将包含 1/10 秒内显示的消息数。我们将这些值称为 bin，每个 bin 保存 1/10 秒的数据。每 100 毫秒，其中一个 bin 就会被丢弃，并且一个新的 bin 会被设置为这 100 毫秒内显示的消息数。

如果您想在整个小时内保持 1/10 秒的精度，则需要一组 36K bin 来保存一小时的有关消息速率的信息。但这似乎有些过分了。

但我认为随着时间间隔变大，精度下降会更合理。

也许您将 1 秒的数据精确到 100 毫秒，将 1 分钟的数据精确到秒，将 1 小时的数据精确到分钟，等等。

Answer 5

我想到维护所有消息的队列，搜索超过一秒的最新消息，并从索引推断计数。然而，这似乎太慢了，而且会占用大量内存。

更好的想法是维护消息的链接列表，将新消息添加到头部（带有时间戳），并在它们过期时从尾部弹出它们。或者甚至不弹出它们 - 只需保留一个指向在所需时间范围内传入的最旧消息的指针，并在该消息过期时将其向头部推进（这允许您使用一个列表跟踪多个时间范围）。

您可以在需要时通过从尾部走到头部来计算计数，或者只是单独存储计数，每当向头部添加值时就递增计数，而每当向尾部前进时就递减计数。