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

Question

我的消息以毫秒级的分辨率进入我的程序(从零到几百条消息,毫秒).

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

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

我不能只保持一个简单的计数,如"最后一秒的1,017条消息",因为我不知道消息何时超过1秒,因此不应该在计数中...

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

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

Answer 1

这是循环缓冲区最容易处理的.

循环缓冲区具有固定数量的元素和指向它的指针.您可以向缓冲区添加元素,并在执行时将指针递增到下一个元素.如果超过固定长度缓冲区,则从头开始.这是一种节省空间和时间的方法来存储"最后N个"项目.

现在,在您的情况下,您可以拥有一个1,000个计数器的循环缓冲区,每个计数器在一毫秒内计算消息数.添加所有1,000个计数器可以为您提供上一秒的总计数.当然,您可以通过逐步更新计数来优化报告部分,即从插入时覆盖的数字中扣除计数,然后添加新数字.

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

这里是C-like伪代码:

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,然后将该时间划分为秒.您可以使用不同的权重运行其中几个以有效地覆盖更长的时间间隔.实际上,您使用的是一个无限长的窗口,因此您不必担心数据过期; 减重为你做的.