异步执行的排队动作/代表

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 dataflow的ActionBlock，并简单地排队一个类，其中包含一个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; 
    } 
} 

一个更简单的办法是使用的ActionActionBlock，但迫使你捕捉到的参数：

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

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