c# - 等待和任务并发

Question

我正在尝试创建一个 TaskScheduler 来限制可以同时运行的线程数。我正在使用 this example .问题是我看到了一些我不理解的行为。

如果我像示例一样创建我的类:

LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(5);
TaskFactory factory = new TaskFactory(lcts);

然后像这样运行它:

foreach (int i = 0; i < 10; ++i) 
{
    factory.StartNew(() => DoWork());
}

工作看起来像这样:

private async Task DoWork()
{
    // Do some stuff here like
    StaticIntValueWorkOne++;

    // And then more stuff that is async here
    int someValue = await DoAdditionalWorkAsync();
    Thread.Sleep(10000);
    StaticIntValueWorkTwo++;
}

我看到的是 StaticIntValueWorkOne 立即增加了 10 次，而 StaticIntValueWorkTwo 只增加了一次。然后在 10 秒后，我看到 StaticIntValueWorkTwo 递增，之后每 10 秒递增一次。我没有得到的是 DoAdditionalWorkAsync() 上的 await 对并发所做的事情。我在想我会看到 StaticIntValueWorkOne 递增一次，然后 StaticIntValueWorkTwo 会递增一次。我错过了什么？我只需要 factor.StartNew() 上的 await 吗？

Answer 1

I am trying to create a TaskScheduler that limits the number of threads that can be run at the same time.

您可能想要 skip straight to the answer . ;)

var scheduler = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, 5)
    .ConcurrentScheduler;

What I'm not getting is what the await on DoAdditionalWorkAsync() is doing with the concurrency.

任务调度程序仅适用于执行 代码。当 async 方法在任务调度程序上执行时，您可以将其视为分解为多个任务，并在每个 await 点中断。默认情况下，在 await 之后，async 方法将重新进入它的任务调度器。 async 方法在 awaiting 时不在任务调度程序中。

因此，当方法正在 awaiting 时，调度限制(一次 5 个)根本不适用。因此，在您的 DoWork 中，该方法将首先递增变量，然后让给任务调度程序。虽然产生了，但它不会“计入”您的并发限制。稍后，当该方法恢复时，它将阻塞线程(执行“计数”)并递增第二个变量。

使用这段代码:

private static void Main(string[] args)
{
    var scheduler = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, 5)
        .ConcurrentScheduler;
    TaskFactory factory = new TaskFactory(scheduler);

    for (int i = 0; i < 10; ++i)
    {
        factory.StartNew(() => DoWork());
    }

    Console.ReadKey();
}

private static int StaticIntValueWorkOne, StaticIntValueWorkTwo;

private static async Task DoWork()
{
    // Do some stuff here like
    Console.WriteLine(DateTime.UtcNow + " StaticIntValueWorkOne" + Interlocked.Increment(ref StaticIntValueWorkOne));

    // And then more stuff that is async here
    await Task.Yield();
    Thread.Sleep(10000);
    Console.WriteLine(DateTime.UtcNow + " StaticIntValueWorkTwo" + Interlocked.Increment(ref StaticIntValueWorkTwo));
}

我得到这个(预期的)输出:

3/20/2015 11:01:53 AM StaticIntValueWorkOne1
3/20/2015 11:01:53 AM StaticIntValueWorkOne5
3/20/2015 11:01:53 AM StaticIntValueWorkOne4
3/20/2015 11:01:53 AM StaticIntValueWorkOne2
3/20/2015 11:01:53 AM StaticIntValueWorkOne3
3/20/2015 11:01:53 AM StaticIntValueWorkOne6
3/20/2015 11:01:53 AM StaticIntValueWorkOne9
3/20/2015 11:01:53 AM StaticIntValueWorkOne10
3/20/2015 11:01:53 AM StaticIntValueWorkOne7
3/20/2015 11:01:53 AM StaticIntValueWorkOne8
3/20/2015 11:02:03 AM StaticIntValueWorkTwo1
3/20/2015 11:02:03 AM StaticIntValueWorkTwo3
3/20/2015 11:02:03 AM StaticIntValueWorkTwo2
3/20/2015 11:02:03 AM StaticIntValueWorkTwo4
3/20/2015 11:02:03 AM StaticIntValueWorkTwo5
3/20/2015 11:02:13 AM StaticIntValueWorkTwo6
3/20/2015 11:02:13 AM StaticIntValueWorkTwo7
3/20/2015 11:02:13 AM StaticIntValueWorkTwo8
3/20/2015 11:02:13 AM StaticIntValueWorkTwo9
3/20/2015 11:02:13 AM StaticIntValueWorkTwo10

如果您想限制异步代码的并发，请查看 SemaphoreSlim 或 TPL Dataflow。