Действия в очереди/делегаты для асинхронного выполнения

Question

Есть ли что-то в рамках, которое позволит мне асинхронно выполнять очередь делегатов?

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

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

Я мог бы написать что-то сам, чтобы сделать это, но если есть что-то встроенное, я мог бы использовать вместо этого, что было бы лучше.

Я кратко рассмотрел ThreadPool.QueueUserWorkItem, так как он позволяет выполнять заказ, но может найти удовлетворительный способ предотвратить более одного исполнения за раз.

Answer 1

Есть ли что-то в рамках, которые позволили бы мне асинхронно выполнять очередь делегатов?

Я бы использовал это как настраиваемый планировщик задач. Затем вы можете поставить очередь и запустить своих делегатов в качестве задач, которые предоставят вам все преимущества обработки исключений, отмены и async/await.

Реализация планировщика задач, который будет выполнять ваши делегаты в последовательном порядке, довольно прост, используя BlockingCollection. SerialTaskScheduler ниже является упрощенной версией Stephen Toub's StaTaskScheduler:

using System; 
using System.Collections.Concurrent; 
using System.Collections.Generic; 
using System.Linq; 
using System.Threading; 
using System.Threading.Tasks; 

namespace Console_21628490 
{ 
    // Test 
    class Program 
    { 
     static async Task DoWorkAsync() 
     { 
      using (var scheduler = new SerialTaskScheduler()) 
      { 
       var tasks = Enumerable.Range(1, 10).Select(i => 
        scheduler.Run(() => 
        { 
         var sleep = 1000/i; 
         Thread.Sleep(sleep); 
         Console.WriteLine("Task #" + i + ", sleep: " + sleep); 
        }, CancellationToken.None)); 

       await Task.WhenAll(tasks); 
      } 
     } 

     static void Main(string[] args) 
     { 
      DoWorkAsync().Wait(); 
      Console.ReadLine(); 
     } 
    } 

    // SerialTaskScheduler 
    public sealed class SerialTaskScheduler : TaskScheduler, IDisposable 
    { 
     Task _schedulerTask; 
     BlockingCollection<Task> _tasks; 
     Thread _schedulerThread; 

     public SerialTaskScheduler() 
     { 
      _tasks = new BlockingCollection<Task>(); 

      _schedulerTask = Task.Run(() => 
      { 
       _schedulerThread = Thread.CurrentThread; 

       foreach (var task in _tasks.GetConsumingEnumerable()) 
        TryExecuteTask(task); 
      }); 
     } 

     protected override void QueueTask(Task task) 
     { 
      _tasks.Add(task); 
     } 

     protected override IEnumerable<Task> GetScheduledTasks() 
     { 
      return _tasks.ToArray(); 
     } 

     protected override bool TryExecuteTaskInline(
      Task task, bool taskWasPreviouslyQueued) 
     { 
      return _schedulerThread == Thread.CurrentThread && 
       TryExecuteTask(task); 
     } 

     public override int MaximumConcurrencyLevel 
     { 
      get { return 1; } 
     } 

     public void Dispose() 
     { 
      if (_schedulerTask != null) 
      { 
       _tasks.CompleteAdding(); 
       _schedulerTask.Wait(); 
       _tasks.Dispose(); 
       _tasks = null; 
       _schedulerTask = null; 
      } 
     } 

     public Task Run(Action action, CancellationToken token) 
     { 
      return Task.Factory.StartNew(action, token, TaskCreationOptions.None, this); 
     } 

     public Task Run(Func<Task> action, CancellationToken token) 
     { 
      return Task.Factory.StartNew(action, token, TaskCreationOptions.None, this).Unwrap(); 
     } 

     public Task<T> Run<T>(Func<Task<T>> action, CancellationToken token) 
     { 
      return Task.Factory.StartNew(action, token, TaskCreationOptions.None, this).Unwrap(); 
     } 
    } 
} 

Выход:

 
Task #1, sleep: 1000 
Task #2, sleep: 500 
Task #3, sleep: 333 
Task #4, sleep: 250 
Task #5, sleep: 200 
Task #6, sleep: 166 
Task #7, sleep: 142 
Task #8, sleep: 125 
Task #9, sleep: 111 
Task #10, sleep: 100 

Answer 2

Вы можете использовать TPL dataflowActionBlock и просто поставить в очередь класс, содержащий Delegate и списки параметров. Команда ActionBlock будет просто выполнять эти делегаты по одному за раз.

var block = new ActionBlock<Item>(_ => _.Action.DynamicInvoke(_.Paramters)); 

class Item 
{ 
    public Delegate Action { get; private set; } 
    public object[] Parameters { get; private set; } 

    public Item(Delegate action, object[] parameters) 
    { 
     Action = action; 
     Parameters = parameters; 
    } 
} 

Еще проще вариант будет использовать ActionBlock из Action, но что заставляет вас захватить параметры:

var block = new ActionBlock<Action>(action => action()); 

block.Post(() => Console.WriteLine(message));