Ограничение количества событий за период времени

Как насчет этих переменных: num_events_allowed, time_before, time_now, time_passed

Во время инициализации вы сделаете: time_before = system.timer(), num_events_allowed = n

Когда событие получено, вы делаете следующее:

  time_now = system.timer()
  time_passed = time_now - time_before
  time_before = time_now

  num_events_allowed += time_passed * (n / deltaT);

  if num_events_allowed > n 
      num_events_allowed = n

  if num_events_allowed >= 1
      let event through, num_events_allowed -= 1
  else
      ignore event

Что хорошо в этом алгоритме, так это то, что num_events_allowed фактически увеличивается на время, прошедшее с момента последнего события, и на скорость получения событий, таким образом, вы получаете приращение количества событий, которые вы можете отправить за это time_passed в порядке оставаться в пределах n. Поэтому, если вы получите событие слишком рано, вы увеличите его менее чем на 1, если через слишком много времени вы увеличите его более чем на единицу. Конечно, если событие проходит, вы уменьшаете допуск на 1, так как вы только что получили событие. Если допуск проходит максимальное количество событий, равное n , вы возвращаете его обратно к n , поскольку вы не можете разрешить более n в любой временной фазе. Если припуск меньше 1, вы не можете отправить целое событие, не пропускайте его!

Это алгоритм дырявого ведра: https://en.wikipedia.org/wiki/Leaky_bucket

Один из способов сохранить скользящее окно и по-прежнему иметь его O (1) + очень маленькое O (n) для каждого входящего запроса — создать массив целых чисел подходящего размера и сохранить его в виде циклического буфера и дискретизировать входящие запросы (запросы как интегрируется с уровнями дискретизации, как в аналого-цифровом преобразователе, или в виде гистограммы, если вы статистик) и следите за суммой кольцевого буфера, как здесь

assumptions: 
"there can be no more than 1000 request per minute" and 
"we discretize on every second"

int[] buffer = new zeroed int-array with 60 zeroes
int request-integrator = 0 (transactional)
int discretizer-integrator = 0 (transactional)

for each request:
    check if request-integrator < 1000 then
         // the following incs could be placed outside 
         // the if statement for saturation on to many
         // requests (punishment)
         request-integrator++                     // important
         discretizer-integrator++
         proceed with request

once every second:                    // in a transactional memory transaction, for God's saké 
    buffer-index++
    if (buffer-index = 60) then buffer-index=0    // for that circular buffer feeling!
    request-integrator -= buffer[buffer-index]    // clean for things happening one minute ago
    buffer[buffer-index] = discretizer-integrator // save this times value
    discretizer-integrator = 0                    // resetting for next sampling period

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

Пока читал о различных возможных решениях проблемы. Я наткнулся на алгоритм ведра токенов ( http://en.wikipedia.org/wiki/Token_bucket) . Если я полностью понимаю ваш вопрос, вы можете реализовать алгоритм корзины токенов, фактически не имея корзины с N токенами, вместо этого используя счетчик, который можно соответственно увеличивать и уменьшать. нравится

syncronized def get_token = 
    if count>0 
       { --count, return true }
    else return false

syncronized def add_token = 
    if count==N
       return;
    else ++count

Теперь осталось повторно вызвать add_token во времени deltaT/r.

Чтобы сделать его полностью потокобезопасным, нам потребуется атомарная ссылка для подсчета. Но приведенный выше код должен показать основную идею выполнения этого в памяти O (1).

Я написал приведенный ниже класс (ActionQueue), чтобы ограничить частоту вызовов функций. Одна из приятных вещей заключается в том, что он использует таймер для извлечения объектов из очереди... поэтому ЦП используется минимально (или даже не используется вообще, если очередь пуста)... в отличие от любого метода опроса. .

Пример...

    // limit to two call every five seconds
    ActionQueue _actionQueue = new ActionQueue(TimeSpan.FromSeconds(5), 2);
    public void Test()
    {
        for (var i = 0; i < 10; i++)
        {
            _actionQueue.Enqueue((i2) =>
            {
                Console.WriteLineAction " + i2 + ": " + DateTime.UtcNow);
            }, i);
        }
    }

Пример реального мира...

    ActionQueue _actionQueue = new ActionQueue(TimeSpan.FromSeconds(1), 10);

    public override void SendOrderCancelRequest(Order order, SessionID sessionID)
    {
        _actionQueue.Enqueue((state) =>
        {
            var parms = (Tuple<Order, SessionID>)state;
            base.SendOrderCancelRequest(parms.Item1, parms.Item2);
        }, new Tuple<Order, SessionID>(order, sessionID));
    }
    public override void SendOrderMassStatusRequest(SessionID sessionID)
    {
        _actionQueue.Enqueue((state) =>
        {
            var sessionID2 = (SessionID)state;
            base.SendOrderMassStatusRequest(sessionID2);
        }, sessionID);
    }

Настоящий класс...

public class ActionQueue
{
    private class ActionState
    {
        public Action<object> Action;
        public object State;
        public ActionState(Action<object> action, object state)
        {
            Action = action;
            State = state;
        }
    }
    Queue<ActionState> _actions = new Queue<ActionState>();
    Queue<DateTime> _times = new Queue<DateTime>();

    TimeSpan _timeSpan;
    int _maxActions;
    public ActionQueue(TimeSpan timeSpan, int maxActions)
    {
        _timeSpan = timeSpan;
        _maxActions = maxActions;           
    }
    public void Enqueue(Action<object> action, object state)
    {
        lock (_times)
        {
            _times.Enqueue(DateTime.UtcNow + _timeSpan);

            if (_times.Count <= _maxActions)
                action(state);
            else
                _actions.Enqueue(new ActionState(action, state));

            CreateDequeueTimerIfNeeded();
        }
    }

    System.Threading.Timer _dequeueTimer;
    protected void CreateDequeueTimerIfNeeded()
    {
        // if we have no timer and we do have times, create a timer
        if (_dequeueTimer == null && _times.Count > 0) 
        {
            var timeSpan = _times.Peek() - DateTime.UtcNow;
            if (timeSpan.TotalSeconds <= 0)
            {
                HandleTimesQueueChange();
            }
            else
            {
                _dequeueTimer = new System.Threading.Timer((obj) =>
                {
                    lock (_times)
                    {
                        _dequeueTimer = null;
                        HandleTimesQueueChange();
                    }
                }, null, timeSpan, System.Threading.Timeout.InfiniteTimeSpan);
            }
        }
    }

    private void HandleTimesQueueChange()
    {
        _times.Dequeue();
        while (_times.Count > 0 && _times.Peek() < DateTime.UtcNow)
            _times.Dequeue();

        while (_actions.Count > 0 && _times.Count < _maxActions)
        {
            _times.Enqueue(DateTime.UtcNow + _timeSpan);
            var actionState = _actions.Dequeue();
            actionState.Action(actionState.State);
        }

        CreateDequeueTimerIfNeeded();
    }
}