Вычисление количества сообщений в секунду в скользящем окне?

Question

В мою программу приходят сообщения с миллисекундным разрешением (от нуля до пары сотен сообщений в миллисекунду).

Я хотел бы провести некоторый анализ.В частности, я хочу поддерживать несколько скользящих окон подсчета сообщений, обновляемых по мере поступления сообщений.Например,

• # сообщений за последнюю секунду
• # сообщений за последнюю минуту
• # сообщений за последние полчаса деленное на # сообщений за последний час

Я не могу просто вести простой подсчет, например «1017 сообщений за последнюю секунду», поскольку я не знаю, когда сообщение старше 1 секунды и, следовательно, больше не должно учитываться...

Я думал о сохранении очереди всех сообщений, поиске самого молодого сообщения старше одной секунды и выводе количества из индекса.Однако кажется, что это будет слишком медленно и съест много памяти.

Что я могу сделать, чтобы отслеживать эти значения в своей программе и эффективно получать эти значения в режиме реального времени?

Answer 1

Проще всего это сделать с помощью циклического буфера.

Циклический буфер имеет фиксированное количество элементов и указатель на него.Вы можете добавить элемент в буфер, и когда вы это сделаете, вы увеличите указатель на следующий элемент.Если вы преодолеете буфер фиксированной длины, вы начнете с самого начала.Это эффективный по пространству и времени способ хранения «последних N» элементов.

Теперь в вашем случае у вас может быть один циклический буфер из 1000 счетчиков, каждый из которых подсчитывает количество сообщений за одну миллисекунду.Если сложить все 1000 счетчиков, вы получите общее количество счетчиков за последнюю секунду.Конечно, вы можете оптимизировать часть отчетности, постепенно обновляя счетчик, т.е.вычтите из счета число, которое вы перезаписываете при вставке, а затем добавьте новый номер.

Затем вы можете иметь еще один циклический буфер с 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 элементов из массива прокручивающегося буфера и поместите их в следующий прокручивающийся буфер.Таким образом, вы можете сохранить небольшой буфер прокрутки миллисекунд (1000 элементов для максимального поиска в течение 1 с), а также буфер для поиска в минуту (600 элементов).

Следующий трюк можно выполнять целыми минутами с интервалом в 0,1 минуты.Все заданные вопросы можно получить путем суммирования (или при использовании cummulative вычитания двух значений) нескольких целых чисел.

Единственным недостатком является то, что значение последней секунды будет меняться каждую мс, а значение минуты - только каждые 0,1 секунды, а значение часа (и производные с % за последние полчаса) каждые 0,1 минуту.Но, по крайней мере, вы сохраняете использование памяти под контролем.

Answer 4

Ваше прокручивающееся окно отображения может обновляться только так быстро, скажем, вы хотите обновлять его 10 раз в секунду, поэтому для данных за 1 секунду вам понадобится 10 значений.Каждое значение будет содержать количество сообщений, появившихся за эту 1/10 секунды.Давайте назовем эти значения интервалами, каждый интервал содержит 1/10 секунды данных.Каждые 100 миллисекунд одна из ячеек отбрасывается, и новая ячейка устанавливается на количество сообщений, появившихся за эти 100 миллисекунд.

Если вы хотите сохранить точность 1/10 секунды в течение всего часа, вам понадобится массив из 36 000 ячеек для хранения информации о частоте ваших сообщений за час.Но это кажется излишним.

Но я думаю, что было бы разумнее снизить точность по мере увеличения временного интервала.

Возможно, вы храните данные за 1 секунду с точностью до 100 миллисекунд, данные за 1 минуту с точностью до секунды, данные за 1 час с точностью до минуты и так далее.

Answer 5

Я думал о сохранении очереди всех сообщений, поиске самого молодого сообщения старше одной секунды и выводе количества из индекса.Однако кажется, что это будет слишком медленно и съест много памяти.

Лучшей идеей было бы поддерживать связанный список сообщений, добавлять новые сообщения в заголовок (с отметкой времени) и удалять их из хвоста по мере истечения срока их действия.Или даже не выталкивать их — просто держите указатель на самое старое сообщение, пришедшее в нужный период времени, и перемещайте его к началу, когда это сообщение истечет (это позволяет отслеживать несколько таймфреймов одним списком).

При необходимости вы можете вычислить счетчик, пройдя от хвоста к началу, или просто сохранить счетчик отдельно, увеличивая его всякий раз, когда вы добавляете значение в начало, и уменьшая его всякий раз, когда вы продвигаете хвост.