Как (и если) написать однопользовательскую очередь с использованием TPL? - PullRequest
Новая
       36

Как (и если) написать однопользовательскую очередь с использованием TPL?

16 голосов
/ 19 февраля 2010

Недавно я слышал несколько подкастов о TPL в .NET 4.0. Большинство из них описывают фоновые действия, такие как загрузка изображений или выполнение вычислений, используя задачи, чтобы работа не мешала потоку GUI.

Большая часть кода, над которым я работаю, имеет более сложный вид с несколькими производителями / одним потребителем, где рабочие элементы из нескольких источников должны быть поставлены в очередь, а затем обработаны по порядку. Одним из примеров может быть ведение журнала, где строки журнала из нескольких потоков секвенируются в одну очередь для возможной записи в файл или базу данных. Все записи из любого отдельного источника должны оставаться в порядке, а записи из одного и того же момента времени должны быть «близки» друг к другу в конечном выводе.

Итак, несколько потоков, задач или чего-то еще вызывают вызов очереди: 

lock( _queue ) // or use a lock-free queue!
{
   _queue.enqueue( some_work );
   _queueSemaphore.Release();
}

И выделенный рабочий поток обрабатывает очередь: 

while( _queueSemaphore.WaitOne() )
{
   lock( _queue )
   {
      some_work = _queue.dequeue();     
   }
   deal_with( some_work );
}

Всегда казалось разумным выделять рабочий поток для потребительской стороны этих задач. Должен ли я писать будущие программы, используя вместо этого какую-то конструкцию из TPL? Который из? Почему?

Ответы [ 4 ]

13 голосов
/ 06 мая 2010

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

Log from task B
Log from task A
Log from task B1
Log from task D
Log from task C

Не то, чтобы выходы из A, B, C & D казались случайными, потому что они зависят от времени начала потоков, но B всегда появляется до B1. 

public class LogItem 
{
    public string Message { get; private set; }

    public LogItem (string message)
    {
        Message = message;
    }
}

public void Example()
{
    BlockingCollection<LogItem> _queue = new BlockingCollection<LogItem>();

    // Start queue listener...
    CancellationTokenSource canceller = new CancellationTokenSource();
    Task listener = Task.Factory.StartNew(() =>
        {
            while (!canceller.Token.IsCancellationRequested)
            {
                LogItem item;
                if (_queue.TryTake(out item))
                    Console.WriteLine(item.Message);
            }
        },
    canceller.Token, 
    TaskCreationOptions.LongRunning,
    TaskScheduler.Default);

    // Add some log messages in parallel...
    Parallel.Invoke(
        () => { _queue.Add(new LogItem("Log from task A")); },
        () => { 
            _queue.Add(new LogItem("Log from task B")); 
            _queue.Add(new LogItem("Log from task B1")); 
        },
        () => { _queue.Add(new LogItem("Log from task C")); },
        () => { _queue.Add(new LogItem("Log from task D")); });

    // Pretend to do other things...
    Thread.Sleep(1000);

    // Shut down the listener...
    canceller.Cancel();
    listener.Wait();
}
5 голосов
/ 09 февраля 2011

Я знаю, что этот ответ опоздал примерно на год, но взгляните на MSDN .

, который показывает, как создать LimitedConcurrencyLevelTaskScheduler из класса TaskScheduler.Ограничивая параллелизм одной задачей, она должна обрабатывать ваши задачи в порядке их очередности: 

LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(1);
TaskFactory factory = new TaskFactory(lcts);

factory.StartNew(()=> 
{
   // your code
});
3 голосов
/ 23 февраля 2010

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

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

ХорошоЭто, конечно, мой личный взгляд на эту тему

Живи долго и процветай

2 голосов
/ 21 марта 2010

Похоже, BlockingCollection будет для вас удобным. Поэтому для приведенного выше кода вы можете использовать что-то вроде (при условии, что _queue является BlockingCollection экземпляром): 

// for your producers 
_queue.Add(some_work);

Выделенный рабочий поток, обрабатывающий очередь: 

foreach (var some_work in _queue.GetConsumingEnumerable())
{
    deal_with(some_work);
}

Примечание: когда все ваши производители закончили производство, вам нужно будет позвонить CompleteAdding() на _queue, иначе ваш потребитель застрянет в ожидании дополнительной работы.

...