0

给定以下代码，是否可以为 的实例定义调度程序、创建和延续设置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;
    }
4

2 回答 2

5

TPL TaskScheduler 一次只能看到一个异步方法的一个同步段，因此您不能简单地使用调度程序来做到这一点。但是您可以使用更高级别的原语来做到这一点。我经常使用的一种是 TPL 数据流。

首先，安装 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 主要不是为跟踪单个提交而设计的，而是为跟踪整个过程而设计的。因此，虽然上述工作正常，但斯蒂芬克利里的答案可能更简单，因此更可取。

于 2013-03-03T17:21:58.477 回答
2

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

坏消息是：不，没有办法做到这一点。为非 lambda 任务定义“调度程序”是没有意义的。不需要创建选项，并且继续选项设置在继续上，而不是在任务本身上。

好消息是：你不需要这种行为。

您需要异步同步。执行此操作的内置方法是使用SemaphoreSlim，如下所示：

SemaphoreSlim mutex = new SemaphoreSlim(1);
private static async Task doThingAsync(object i)
{
    await mutex.WaitAsync();
    try
    {
        Console.WriteLine("in do thing {0}", (int)i);
        await Task.Delay(TimeSpan.FromSeconds(5));
        Console.WriteLine("out of do thing {0}", (int)i);
    }
    finally
    {
        mutex.Release();
    }
}

就个人而言，我认为finally语法很尴尬，所以我定义了一个IDisposable并使用using。

如果您需要更多功能，Stephen Toub 有一个异步协调原语系列，我的 AsyncEx 库中有一整套原语。这两个资源都包含一个AsyncLockwithTask<IDisposable> WaitAsync()成员，因此您可以使用using而不是finally.

于 2013-03-01T22:08:25.650 回答