Наиболее эффективная блокировка для многих писателей Модель параллельного чтения с одним читателем?

Question

Итак, у меня много потоков, которые подают мне входные данные, которые должны обрабатываться одним потоком в порядке поступления.В настоящее время все элементы ввода вставляются в очередь, а чтение / запись в очередь защищена оператором блокировки C #.Однако со временем загрузка ЦП приложения возрастает до неприемлемого уровня, и профилировщик говорит, что большая часть времени ЦП расходуется на сам оператор блокировки.Есть ли более эффективный метод синхронизации вместо блокировки, который поддерживает много писателей и одного читателя?

Answer 1

Звучит так, будто писатели борются друг с другом за замки.Рассмотрим модель, в которой у каждого средства записи есть своя собственная очередь, и в которой читатель использует метод Peek , чтобы прочитать первое сообщение из каждой очереди, не удаляя его.Затем читатель может продолжать итерации между очередями, просматривая первый элемент среди набора первых элементов из каждой очереди, а затем удаляя и обрабатывая этот первый элемент.Это будет медленнее, чем ваша текущая архитектура, но должно устранить конфликт блокировки среди писателей.

Тривиальный пример может выглядеть так:

public class TimestampedItem<T> : IComparable<TimestampedItem<T>>
{
    public DateTime TimeStamp { get; set; }
    public T Data { get; set; }
    public int CompareTo(TimestampedItem<T> other)
    {
        return TimeStamp.CompareTo(other.TimeStamp);
    }
}

public void ReadFirstFromEachQueue<T>(IEnumerable<Queue<TimestampedItem<T>>> queues)
{
    while (true)
    {
        var firstItems = new List<TimestampedItem<T>>(queues.Select(q => { lock (q) { return q.Peek(); } }));
            ProcessItem(firstItems.OrderBy(tsi => tsi.TimeStamp).First());
        }
    }
}

Answer 2

Если вы используете .net версии 4.0, вы можете использовать ConcurrentQueue, который является частью ConcurrentCollections, вместо обычного Queue, а затем избавиться от блокировки при чтении / записи ваших данных.ConcurrentCollections

Если вы не используете 4.0, вы можете заблокировать, только если никакая другая блокировка не установлена ​​в очередь, очередь *1005* предназначена для обработки одновременного чтения / записи без кода блокировки.но вы можете добиться этого, используя Monitor.TryEnter вместо lock, отметив, что lock само по себе является комбинацией Monitor.Enter и Monitor.Exit. Пример реализации будет:

Answer 3

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

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