Обработка очереди элементов асинхронно в C #

Question

Я пытаюсь создать систему, которая обрабатывает очередь работы. Система имеет следующие технические характеристики:

1. Система состоит из двух компонентов: рабочего задания и рабочего.
2. Существует установленный верхний предел количества работающих одновременно работников. Этот верхний предел больше единицы.
3. Во избежание проблем, связанных с выполнением одной и той же задачи дважды, существует только один назначитель работы.

Какой дизайн вы бы использовали для создания такой системы? Вот что я думаю:

1. Создать коллекцию очередей, по одной очереди на каждого работника
2. Создать таймер для рабочего задания. Его работа заключается в заполнении очередей.
3. Создать таймер для каждого работника, передавая объект очереди в качестве состояния объекта для представления его рабочей нагрузки
4. Удалить и добавить в очереди, пока они заблокированы.
5. Используйте счетчик, который увеличивается и уменьшается при блокировке, чтобы гарантировать, что одновременно выполняется не более указанного числа рабочих задач.

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

Answer 1

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

Что касается обеспечения безопасности потока в очереди, в System.Collections.Concurrent есть ConcurrentQueue для этой цели ( msdn , тест по сравнению с очередью блокировки ).

Теперь добавьте BlockingCollection ( msdn ), и у вас есть все, что вам нужно.

        BlockingCollection<Packet> sendQueue = new BlockingCollection<Packet>(new ConcurrentQueue<Packet>());
        while (true)
        {
            var packet = sendQueue.Take(); //this blocks if there are no items in the queue.
            ThreadPool.QueueUserWorkItem(state =>
            {
               var data = (Packet)state;
               //do whatever you have to do
            }, packet );
        }

и где-то есть что-то, что sendQueue.Add(packet);

Подводя итог,

1. Одна очередь для всех «рабочих»
2. Один поток, который исключает из очереди и передает его в ThreadPool.

Я думаю, что это так.

ps: если вам нужно контролировать количество потоков, используйте «Smart Thread Pool», как предложено josh3736

Answer 2

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

Answer 3

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