c# - 异步执行的排队操作/委托(delegate)

Question

框架中是否有允许我异步执行委托(delegate)队列的东西？

我的意思是我希望委托(delegate)按照他们排队的顺序一次执行一个，但我希望整个过程异步运行。队列也不固定，会定期添加额外的委托(delegate)，并应在到达队列顶部后立即处理。

我不需要特别使用 Queue，这只是我描述所需行为的方式。

我可以自己写一些东西来做，但如果有内置的东西我可以改用那会更好。

我简要地查看了 ThreadPool.QueueUserWorkItem，因为它允许按顺序执行，但可以找到一种令人满意的方法来防止一次执行多个。

Answer 1

Is there something in the framework that would allow me to asynchronously execute a queue of delegates?

我会将其实现为自定义任务调度程序。然后，您可以将委托(delegate)作为任务排队并运行，这将为您提供异常处理、取消和 async/await 的所有好处。

使用 BlockingCollection 实现一个按顺序执行委托(delegate)的任务调度程序非常简单。下面的 SerialTask​​Scheduler 是 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: 1000Task #2, sleep: 500Task #3, sleep: 333Task #4, sleep: 250Task #5, sleep: 200Task #6, sleep: 166Task #7, sleep: 142Task #8, sleep: 125Task #9, sleep: 111Task #10, sleep: 100