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

Question

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

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

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

Я не могу просто вести простой подсчет, такой как "1,017 сообщений в последнюю секунду" , так как я не буду знать, когда сообщение старше 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 минуты. Все задаваемые вопросы могут быть запрошены суммированием (или при использовании кумулятивного, вычитания двух значений) нескольких целых чисел.

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

Answer 4

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

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

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

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

Answer 5

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

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

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