Как сделать потокобезопасный обработчик событий

Question

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

Ниже приведен код для обработчика событий:

private async void NotificationQueue_Changed(object sender, EventArgs e) 
{ 
    if (!IsQueueInProcess) 
    await ProcessQeueue(); 
} 

в ProcessQueue методы я устанавливаю IsQueueInProcess истину и всякий раз, когда она будет завершен его установлено значение false. Теперь проблема заключается в том, что всякий раз, когда несколько уведомлений о событиях срабатывают одновременно, запускаются несколько методов ProcessQeueue, чего я не хочу. Я хочу убедиться, что будет только одно исполнение ProcessQeueue в любой момент времени.

Answer 1

Учитывая ваше утверждение, что это событие возникает всякий раз, когда есть какие-либо изменения в очереди, и что очередь может использоваться одновременно (т. Е. Есть несколько продюсеров, добавляющих вещи в очередь), мне кажется, что лучший способом решения этой проблемы было бы вообще отказаться от поведения, основанного на событиях. Вместо этого, используя BlockingCollection<T>, с потоком, предназначенным для обработки очереди через GetConsumingEnumerable(). Этот метод блокирует поток, пока очередь пуста, и позволит потоку удалять и обрабатывать элементы в очереди в любое время, когда какой-либо другой поток добавляет что-то к нему. Сама коллекция является потокобезопасной, поэтому, если вы не потребуете дополнительной синхронизации потоков (для обработки очереди, которая является & hellip; возможно, обработка элемента связана с взаимодействием потоков, но в вашем вопросе нет ничего, что описывает этот аспект, поэтому я не могу сказать об этом никому.

Тем не менее, принимая вопрос буквально, самый простой подход будет включать в себя семафор:

private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1); 

private async void NotificationQueue_Changed(object sender, EventArgs e) 
{ 
    if (_semaphore.Wait(0)) 
    { 
     await ProcessQueue(); 
     _semaphore.Release(); 
    } 
} 

Вышеуказанные попытки приобрести блокировку семафора. При тайм-ауте в 0 миллисекунд он немедленно возвращается, даже если семафор не может быть приобретен. Возвращаемое значение указывает, был ли семафор успешно приобретен или нет.

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

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

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

В вашем вопросе недостаточно контекста, чтобы кто-нибудь мог решить эту проблему конкретно. Я просто укажу, что это достаточно распространенная вещь, которую кто-то может игнорировать при попытке реализовать такую ​​систему.IMHO, тем более, что для выделения элементов, добавленных в очередь, требуется только выделенный поток с использованием BlockingCollection<T>. :)

См. Также соответствующий вопрос How to avoid reentrancy with async void event handlers?. Это немного другой вопрос, поскольку принятый ответ заставляет каждый вызов обработчика события приводить к операции, инициированной обработчиком события. Ваш сценарий проще, так как вы просто хотите пропустить начало новой операции, но вы все равно можете найти там полезную информацию.

Answer 2

Я согласен с Питером в том, что отказ от уведомлений, основанных на событиях, является лучшим решением и что вы должны перейти в очередь производителей/потребителей. Однако я рекомендую использовать один из блоков потока данных TPL вместо BlockingCollection<T>.

В частности, ActionBlock<T> должен работать достаточно хорошо:

private readonly ActionBlock<T> notificationQueue = new ActionBlock<T>(async t => 
{ 
    await ProcessQueueItem(t); 
}); 

По умолчанию TPL DataFlow блоки имеют предел параллелизм 1.