Блог

Я пытаюсь реализовать довольно простое приложение в стиле производителя / потребителя с несколькими производителями и одним потребителем.

Исследование привело меня к BlockingCollection<T> тому, что является полезным и позволило мне реализовать долгосрочную задачу потребителя, как показано ниже:

 var c1 = Task.Factory.StartNew(() =>
{
    var buffer = new List<int>(BATCH_BUFFER_SIZE);

    foreach (var value in blockingCollection.GetConsumingEnumerable())
    {
        buffer.Add(value);
        if (buffer.Count == BATCH_BUFFER_SIZE)
        {
            ProcessItems(buffer);
            buffer.Clear();
        }
    }
});
 

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

Более идеальным решением было бы либо запустить задачу по 30-секундному таймеру, либо замкнуть foreach ее с таймаутом.

Я работал с идеей таймера и придумал это:

 syncTimer = new Timer(new TimerCallback(TimerElapsed), blockingCollection, 5000, 5000);

private static void TimerElapsed(object state)
{
    var buffer = new List<int>();
    var collection = ((BlockingCollection<int>)state).GetConsumingEnumerable();

    foreach (var value in collection)
    {
        buffer.Add(value);
    }

    ProcessItems(buffer);
    buffer.Clear();
}
 

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

Кто-нибудь может предложить направление? По сути, мне нужно BlockingCollection периодически делать снимок и обрабатывать содержимое, чтобы очистить его. Возможно, a BlockingCollection — неправильный тип?

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

Блокирующая коллекция.Метод TryTake (T) — вероятно, то, что вам нужно, вы вообще не хотите выполнять дальнейшее ожидание.

Блокирующая коллекция.Метод TryTake (T, Int32)

Блокирующая коллекция.Метод TryTake (T, TimeSpan)

Вы можете легко извлечь это в расширение (непроверенное):

 public static IList<T> Flush<T>
(this BlockingCollection<T> collection, int maxSize = int.MaxValue)
{
     // Argument checking.

     T next;
     var result = new List<T>();

     while(result.Count < maxSize amp;amp; collection.TryTake(out next))
     {
         result.Add(next);
     }

     return resu<
}