如何按顺序执行任意数量的异步任务？

Question

我有这个功能：

async Task RefreshProfileInfo(List<string> listOfPlayers)

// For each player in the listOfPlayers, checks an in-memory cache if we have an entry.
// If we have a cached entry, do nothing.
// If we don't have a cached entry, fetch from backend via an API call.

这个函数调用非常频繁，比如：

await RefreshProfileInfo(playerA, playerB, playerC)

或

await RefreshProfileInfo(playerB, playerC, playerD) 

或

await RefreshProfileInfo(playerE, playerF)

理想情况下，如果玩家不相互重叠，则调用不应相互影响（请求 PlayerE 和 PlayerF 不应阻止对 PlayerA、PlayerB、PlayerC 的请求）。但是，如果玩家确实相互重叠，则第二个调用应该等待第一个调用（请求 PlayerB、PlayerC、PlayerD，应该等待 PlayerA、PlayerB、PlayerC 完成）。

但是，如果这不可能，至少我希望所有调用都是连续的。 （我认为它们仍然应该是异步的，因此它们不会阻塞 other 代码的不相关部分）。

目前，发生的情况是每个 RefreshProfileInfo 并行运行，这导致每次都命中后端（本例中为 9 次）。

相反，我想按顺序执行它们，这样只有第一个调用会命中后端，后续调用只会命中缓存。

我应该使用什么数据结构/方法？我无法弄清楚如何将单独的呼叫“连接”到彼此。我一直在玩 Task.WhenAll() 和 SemaphoreSlim，但我不知道如何正确使用它们。

尝试失败

我失败的尝试背后的想法是有一个帮助类，我可以在其中调用一个函数 SequentialRequest(Task)，它会按顺序运行以这种方式调用的所有任务。

List<Task> waitingTasks = new List<Task>();
object _lock = new object();

public async Task SequentialRequest(Task func)
{
    var waitingTasksCopy = new List<Task>();

    lock (_lock)
    {
        waitingTasksCopy = new List<Task>(waitingTasks);
        waitingTasks.Add(func); // Add this task to the waitingTasks (for future SequentialRequests)
    }

    // Wait for everything before this to finish
    if (waitingTasksCopy.Count > 0)
    {
        await Task.WhenAll(waitingTasksCopy);
    }

    // Run this task
    await func;
}

我认为这会起作用，但“func”要么立即运行（而不是等待早期任务完成），要么根本不运行，这取决于我如何称呼它。

如果我用这个来调用它，它会立即运行：

async Task testTask()
{
    await Task.Delay(4000);
}

如果我用这个来调用它，它永远不会运行：

Task testTask = new Task(async () =>
{
    await Task.Delay(4000);
});

Answer 1

这就是您当前的尝试不起作用的原因：

// Run this task
await func;

上面的注释没有描述代码在做什么。在异步世界中，Task 代表一些操作已经在进行中。使用await 不会“运行”任务； await 它是当前代码“异步等待”任务完成的一种方式。因此，没有采用Task 的函数签名会起作用；该任务甚至在传递给该函数之前就已经在进行中。

您的问题实际上是关于缓存异步操作。一种方法是缓存Task<T> 本身。目前，您的缓存保存结果 (T)；您可以更改缓存以保存检索这些结果的异步操作 (Task<T>)。例如，如果您当前的缓存类型为ConcurrentDictionary<PlayerId, Player>，则可以将其更改为ConcurrentDictionary<PlayerId, Task<Player>>。

对于任务缓存，当您的代码检查缓存条目时，如果播放器数据已加载或已开始加载，它将找到现有条目。因为Task<T> 代表一些已经在进行中（或已经完成）的异步操作。

关于这种方法的几点说明：

1. 这仅适用于内存缓存。
2. 想想你想如何处理错误。 Task<T> 的原始缓存也会缓存 error 结果，这通常是不希望的。

上面的第二点是比较棘手的部分。当错误发生时，您可能需要一些额外的逻辑来删除缓存中的错误任务。如果错误处理代码首先阻止错误任务进入缓存，则加分（以及额外的复杂性）。

至少我希望所有调用都是连续的

嗯，这要容易得多。 SemaphoreSlim 是 lock 的异步替换，因此您可以使用共享的 SemaphoreSlim。在RefreshProfileInfo 的开头调用await mySemaphoreSlim.WaitAsync();，将正文放在try 中，在RefreshProfileInfo 末尾的finally 块中，调用mySemaphoreSlim.Release();。这会将所有对RefreshProfileInfo 的调用限制为按顺序运行。

Answer 2

我在我的一个项目中遇到了同样的问题。我有多个线程调用一个方法，当在缓存中找不到时，它们都进行了 IO 调用。您要做的是将Task 添加到您的缓存中，然后添加await 它。任务完成后，后续调用将只读取结果。

例子：

private Task RefreshProfile(Player player)
{
     // cache is of type IMemoryCache
     return _cache.GetOrCreate(player, entry => 
     {
        // expire in 30 seconds
        entry.AbsoluteExpiration = DateTimeOffset.UtcNow.AddSeconds(30);
        return ActualRefreshCodeThatReturnsTask(player);
     });
}

然后只需在您的调用代码中添加await

await Task.WhenAll(RefreshProfile(Player a), RefreshProfile(Player b), RefreshProfile(Player c));