如何让多个线程等待一个任务？

Question

我读过这篇文章：Is it ok to await the same task from multiple threads - is await thread safe?，但我并不清楚答案，所以这里有一个具体的用例。

我有一个方法可以执行一些异步网络 I/O。多个线程可以同时使用此方法，我不希望它们都调用网络请求，如果请求已经在进行中，我想阻塞/等待第二个线程，并让它们在单个 IO 操作后全部恢复已完成。

应该我如何编写以下伪代码？ 我猜每个调用线程真的需要得到自己的Task，所以每个线程都可以得到它自己的延续，所以不是返回currentTask，我应该返回一个新的Task，它由“内部”@987654327完成@来自DoAsyncNetworkIO。 有没有一种干净的方法可以做到这一点，还是我必须手动滚动它？

static object mutex = new object();
static Task currentTask;

async Task Fetch()
{
    lock(mutex)
    {
        if(currentTask != null)
            return currentTask;
    }

    currentTask = DoAsyncNetworkIO();
    await currentTask;

    lock(mutex)
    {
        var task = currentTask;
        currentTask = null;
        return task;
    }
}

Answer 1

您可以使用SemaphoreSlim 来确保只有一个线程实际执行后台线程。

假设您的基本任务（实际执行 IO 的任务）在一个名为 baseTask() 的方法中，我将模拟如下：

static async Task baseTask()
{
    Console.WriteLine("Starting long method.");
    await Task.Delay(1000);
    Console.WriteLine("Finished long method.");
}

然后你可以像这样初始化SemaphoreSlim，有点像AutoResetEvent，初始状态设置为true：

static readonly SemaphoreSlim signal = new SemaphoreSlim(1, 1);

然后将对baseTask() 的调用封装在一个检查signal 的方法中，以查看这是否是第一个尝试运行baseTask() 的线程，如下所示：

static async Task<bool> taskWrapper()
{
    bool firstIn = await signal.WaitAsync(0);

    if (firstIn)
    {
        await baseTask();
        signal.Release();
    }
    else
    {
        await signal.WaitAsync();
        signal.Release();
    }

    return firstIn;
}

那么你的多个线程将等待taskWrapper()，而不是直接等待baseTask()。

将其完全放入可编译的控制台应用程序中：

using System;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    static class Program
    {
        static void Main()
        {
            for (int it = 0; it < 10; ++it)
            {
                Console.WriteLine($"\nStarting iteration {it}");
                Task[] tasks = new Task[5];

                for (int i = 0; i < 5; ++i)
                    tasks[i] = Task.Run(demoTask);

                Task.WaitAll(tasks);
            }

            Console.WriteLine("\nFinished");                  
            Console.ReadLine();
        }

        static async Task demoTask()
        {
            int id = Thread.CurrentThread.ManagedThreadId;
            Console.WriteLine($"Thread {id} starting");

            bool firstIn = await taskWrapper();

            Console.WriteLine($"Task {id}: executed: {firstIn}");
        }

        static async Task<bool> taskWrapper()
        {
            bool firstIn = await signal.WaitAsync(0);

            if (firstIn)
            {
                await baseTask();
                signal.Release();
            }
            else
            {
                await signal.WaitAsync();
                signal.Release();
            }

            return firstIn;
        }

        static async Task baseTask()
        {
            Console.WriteLine("Starting long method.");
            await Task.Delay(1000);
            Console.WriteLine("Finished long method.");
        }

        static readonly SemaphoreSlim signal = new SemaphoreSlim(1, 1);
    }
}

（这些方法都是静态的，因为它们在控制台应用程序中；在实际代码中它们将是非静态方法。）

Answer 2

await 根本不需要使用延续（Task.ContinueWith 类型）。即使是这样，您也可以在一个 Task 上拥有多个延续 - 它们只是不能全部同步运行（如果您有同步上下文，您可能会遇到一些问题）。

请注意，您的伪代码不是线程安全的 - 您不能只在锁之外执行 currentTask = DoAsyncNetworkIO();。只有await 本身是线程安全的，即使那样，只是因为您正在等待的Task 类以线程安全的方式实现await 合约。任何人都可以编写自己的 awaiter/awaitable，所以请务必注意 :)