如何安排任务以使它们不同时运行,包括异步延续

Question

给定以下代码,是否可以为实例定义调度程序,创建和继续设置Task doThing？

我希望能够安排多个实例,doThing以便它们实际上只从其他实例运行(即使它们正在等待其他子任务).

    private static async Task doThing(object i)
    {
        Console.WriteLine("in do thing {0}", (int)i);
        await Task.Delay(TimeSpan.FromSeconds(5));
        Console.WriteLine("out of do thing {0}", (int)i);
    }
    static void Main(string[] args)
    {

        CancellationTokenSource source = new CancellationTokenSource();
        ConcurrentExclusiveSchedulerPair pair = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Current);

        Task Task1 = Task.Factory.StartNew((Func<object,Task>)doThing, 1, source.Token, TaskCreationOptions.AttachedToParent, pair.ExclusiveScheduler).Unwrap();
        Task Task2 = Task.Factory.StartNew((Func<object, Task>)doThing, 2, source.Token, TaskCreationOptions.AttachedToParent, pair.ExclusiveScheduler);
        Task Task3 = doThing(3);
        Task Task4 = Task.Factory.StartNew(async (i) =>
        {
            Console.WriteLine("in do thing {0}", (int)i);
            await Task.Delay(TimeSpan.FromSeconds(5));
            Console.WriteLine("out of do thing {0}", (int)i);
        }, 4, source.Token, TaskCreationOptions.None, pair.ExclusiveScheduler);
        Task.WaitAll(Task1, Task2, Task3, Task4);
        Console.ReadKey();
        return;
    }

Answer 1

TPL TaskSchedulers一次只能看到异步方法的一个同步段,因此您不能仅使用调度程序来完成.但你可以使用更高级别的原语来做到这一点.我经常使用的是TPL Dataflow.

首先,安装NuGet包:

Install-Package Microsoft.Tpl.Dataflow

然后使用此代码:

private static async Task doThing(object i) {
    Console.WriteLine("in do thing {0}", (int)i);
    await Task.Delay(TimeSpan.FromSeconds(5));
    Console.WriteLine("out of do thing {0}", (int)i);
}

static void Main(string[] args) {
    CancellationTokenSource source = new CancellationTokenSource();
    var exclusivityBlock = new ActionBlock<Func<Task>>(f => f(), new ExecutionDataflowBlockOptions { CancellationToken = source.Token }};
    exclusivityBlock.Post(() => doThing(1));
    exclusivityBlock.Post(() => doThing(2));
    exclusivityBlock.Post(() => doThing(3));
    exclusivityBlock.Post(
        async () => {
            Console.WriteLine("in do thing {0}", 4);
            await Task.Delay(TimeSpan.FromSeconds(5));
            Console.WriteLine("out of do thing {0}", 4);
        });
    exclusivityBlock.Complete();
    exclusivityBlock.Completion.Wait();
    Console.WriteLine("Done");
    Console.ReadKey();
    return;
}

此代码缺少每个已发布工作项的单个任务.如果这很重要,您可以使用此示例:

internal static class Program {
    private static async Task doThing(object i) {
        Console.WriteLine("in do thing {0}", (int)i);
        await Task.Delay(TimeSpan.FromSeconds(5));
        Console.WriteLine("out of do thing {0}", (int)i);
    }

    private static void Main(string[] args) {
        CancellationTokenSource source = new CancellationTokenSource();
        var exclusivityBlock = CreateTrackingBlock<Func<Task>>(
            f => f(), new ExecutionDataflowBlockOptions { CancellationToken = source.Token });
        var task1 = exclusivityBlock.PostWithCompletion(() => doThing(1));
        var task2 = exclusivityBlock.PostWithCompletion(() => doThing(2));
        var task3 = exclusivityBlock.PostWithCompletion(() => doThing(3));
        var task4 = exclusivityBlock.PostWithCompletion(
            async () => {
                Console.WriteLine("in do thing {0}", 4);
                await Task.Delay(TimeSpan.FromSeconds(5));
                Console.WriteLine("out of do thing {0}", 4);
            });

        Task.WaitAll(task1, task2, task3, task4);
        Console.WriteLine("Done");
        Console.ReadKey();
        return;
    }

    private static ActionBlock<Tuple<T, TaskCompletionSource<object>>> CreateTrackingBlock<T>(Func<T, Task> action, ExecutionDataflowBlockOptions options = null) {
        return new ActionBlock<Tuple<T, TaskCompletionSource<object>>>(
            async tuple => {
                try {
                    await action(tuple.Item1);
                    tuple.Item2.TrySetResult(null);
                } catch (Exception ex) {
                    tuple.Item2.TrySetException(ex);
                }
            },
            options ?? new ExecutionDataflowBlockOptions());
    }

    internal static Task PostWithCompletion<T>(this ActionBlock<Tuple<T, TaskCompletionSource<object>>> block, T value) {
        var tcs = new TaskCompletionSource<object>();
        var tuple = Tuple.Create(value, tcs);
        block.Post(tuple);
        return tcs.Task;
    }
}

但请注意,这只是一点费力,因为Dataflow主要不是为跟踪个别提交而是为了跟踪整个流程.因此,尽管上述工作正常,但Stephen Cleary的回答可能更简单,因此更为可取.