如何强制 C# 任务对于给定的 ID 一次只存在一次

Question

在不同的应用程序中，我多次需要使用 C# Task 完成以下行为，并且我以某种方式完成了它，并且希望了解它是否是实现预期效果的最佳方式，或者是否存在是其他更好的方法。

问题在于，在某些情况下，我希望特定任务仅存在于一个实例中。例如，如果有人请求，假设一个产品列表通过执行类似Task GetProductsAsync() 的方法，而其他人尝试请求相同的东西，它不会触发另一个任务，而是返回已经存在的任务。当GetProductsAsync 结束时，之前请求结果的所有调用者都将收到相同的结果。因此，在给定时间点应该只有一次 GetProductsAsync 执行。

在尝试找到类似且众所周知的设计模式来解决此问题后，我想出了自己的实现。在这里

public class TaskManager : ITaskManager
    {
        private readonly object _taskLocker = new object();
        private readonly Dictionary<string, Task> _tasks = new Dictionary<string, Task>();
        private readonly Dictionary<string, Task> _continuations = new Dictionary<string, Task>();

        public Task<T> ExecuteOnceAsync<T>(string taskId, Func<Task<T>> taskFactory)
        {
            lock(_taskLocker)
            {

                if(_tasks.TryGetValue(taskId, out Task task))
                {
                    if(!(task is Task<T> concreteTask))
                    {
                        throw new TaskManagerException($"Task with id {taskId} already exists but it has a different type {task.GetType()}. {typeof(Task<T>)} was expected");
                    }
                    else
                    {
                        return concreteTask;
                    }
                }
                else
                {
                    Task<T> concreteTask = taskFactory();
                    _tasks.Add(taskId, concreteTask);
                    _continuations.Add(taskId, concreteTask.ContinueWith(_ => RemoveTask(taskId)));
                    return concreteTask;
                }
            }
        }

        private void RemoveTask(string taskId)
        {
            lock(_taskLocker)
            {
                if(_tasks.ContainsKey(taskId))
                {
                    _tasks.Remove(taskId);
                }

                if(_continuations.ContainsKey(taskId))
                {
                    _continuations.Remove(taskId);
                }
            }
        }
    }

我们的想法是在整个应用程序生命周期中我们将拥有一个 TaskManager 实例。任何应在给定时间点仅执行一次的异步任务请求将调用ExecuteOnceAsync，提供创建任务本身的工厂方法，以及所需的应用程序范围的唯一 ID。任何其他具有相同 ID 的任务，任务管理器都会回复之前创建的相同任务实例。只有当没有其他具有该 ID 的任务时，管理器才会调用工厂方法并启动任务。我在代码任务创建和删除周围添加了锁，以确保线程安全。此外，为了在任务完成后从存储的字典中删除任务，我使用ContinueWith 方法添加了一个延续任务。因此，在任务完成后，任务本身及其延续都将被删除。

在我看来，这似乎是一个很常见的情况。我会假设有一个完善的设计模式，或者可能是 C# API 可以完成同样的事情。因此，任何见解或建议将不胜感激。

Answer 1

如果您想允许具有相同taskId 但数据类型不同的多个任务同时运行，那么您可以将泛型类型<T> 从方法移动到类。这样每个类型<T> 都有自己的字典。它还允许您使字典存储Task<T> 并避免类型转换。

您说过您将只有一个 TaskManager 类的实例，但如果您可以将其更改为每个调用 TaskManager 的类只有一个实例，您就可以避免需要字典 (并因此散列任务的名称）每个类都有一个

的实例

public class TaskManager<T>
{
    private Task<T> _currentTask;
    private object _lock = new object();

    public Task<T> ExecuteOnceAsync(string taskId, Func<Task<T>> taskFactory)
    {
        if (_currentTask == null)
        {
            lock (_lock)
            {
                if (_currentTask == null)
                {
                    Task<T> concreteTask = taskFactory();
                    concreteTask.ContinueWith(RemoveTask);
                    _currentTask = concreteTask;
                }
            }
        }

        return _currentTask;
    }

    private void RemoveTask()
    {
        _currentTask = null;
    }
}

如果你想超级正确，那么你可以使用Interlocked.Exchange而不是直接给_currentTask赋值，这在传递给ContinueWith的匿名函数中更重要，因为它在锁之外运行。您可能想考虑使用一些Interlocked 方法来检查该值是否为空。顺便说一句，如果您知道对 ExecuteOnceAsync 的调用不会并行发生，则不需要锁定，例如，如果它被 WPF 或 WinForms 应用程序中的按钮单击事件处理程序调用（即使用户单击多个次，每个都应该按顺序发生）。但是，Web 应用程序无法做出这样的保证。

你可以看到我做了两次if (_currentTask == null) 检查，一次在锁内，一次在锁外，这允许任务已经存在的情况，以避免需要获取锁的性能命中。我不相信您可以对问题中的代码使用相同的技巧，因为如果 RemoveTask 与 TryGetValue 同时被调用，您可能会陷入糟糕的状态。您可以切换到使用 ConcurrentDictionary，这可能会让您避免使用 RemoveTask 内部的锁，并且仅在 TryGetValue 返回 false 时锁定 ExecuteOnceAsync。但这是否会提高性能需要衡量。

您和我的代码的另一个问题是，当taskFactory 很慢时，它会长时间保持锁定（请记住，在调用异步方法时，it runs synchronously until the first await where the task it awaits is not already complete）。这可以通过使用Task<T> concreteTask = Task.Run(()=> taskFactory()); 来缓解，但是Task.Run 有它自己的开销，所以如果taskFactory 确实非常快地执行await，那么保持原样可能会更好的性能。同样，如果性能很重要，您需要衡量。

虽然我绝不是性能专家，但根据您的性能要求，我的建议有几个改进：

• 虽然Dictionary 的查找时间为 O(1)，Thread.Sleep(1000) 的查找时间也是如此，换句话说，它可能总是比替代方案慢，即使替代方案是 O(n)（对于较小的 n 值）。由于您的字典键是一个字符串，string.GetHashCode 是 O(n)，具体取决于字符串的长度，并且字典性能可能会受到哈希冲突的影响。我的建议只有一个任务对象，所以它也是 O(1)，它不需要散列任何东西或搜索数组，或者实现 Dictionary。
• 它非常小，但由于我的 RemoveTask 不需要捕获任何变量，我可以将它作为委托直接传递给 ContinueWith（因为类型匹配），而您使用匿名函数，这意味着编译器必须创建一个匿名类来捕获taskId变量，并在每次代码运行时在堆上创建该类的实例，这会增加内存压力。这本身不会产生任何显着的性能影响（除非在非常紧密的循环中运行），但如果您有许多其他代码在堆上创建小对象，它可能会导致“1000 次削减”。
• 如上所述，当任务已经存在时，我的代码是无锁的。
• 您的代码有效地使用了全局应用程序锁。 这是最重要的性能杀手。如果每次发出数据库请求时都使用您的TaskManager，那么您的实现一次只能启动一个数据库查询，即使具有不同taskId 的两个任务永远不会返回相同的任务。而RemoveTask，无论哪个任务完成，都会阻止对ExecuteOnceAsync 的所有调用。通过对每个查询类型使用不同的 TaskManager 实例，您可以避免 GetProductsAsync 阻止 GetCustomerInformation。

写完所有这些，如果您的目标不是每个应用程序使用一个TaskManager，而是每个数据库查询方法使用一个TaskManager，这意味着taskId实际上是@987654361的查询参数的表示@，那么您应该忽略我写的所有内容:)好吧，几乎所有内容。您仍然可以将 <T> 移动到类和字典定义中，并避免在您的方法中进行类型转换。

Answer 2

我认为您可以使用一个有用的并发类来做到这一点。

只有在无法获得任何现有工作时才开始工作，然后等待它完成。如果之前已经运行过，它将获取已经完成（或正在进行）的任务并等待它完成。

// Add this static dictionary to your class

static readonly ConcurrentDictionary<string, Task> tasks = new();

// Add this to your doing something once method in that class

var work = this.tasks.GetOrAdd(taskId, _ =>
{
    return client.DoSomethingAsync();
});

await work;

重要

请参阅下面 Theodor 关于在锁外运行的委托的评论。这很不幸，这可以追溯到我 10 多年前设计的并发字典（并被微软认为是一种更好的方法来做 ConcurrentDictionary）。

我有自己的使用这种方法的线程安全字典。

bool TryAdd(K key, T value, out T contains);

这种简单的设计返回新添加的或现有的值，并提供了一些很棒的模式，特别是对于缓存/请求重复数据删除；添加任务以获得某些东西与等待现有任务获得它。

我注意到ConcurrentDictionary 有一个TryAdd，但它不会在失败时返回现有值，这很遗憾，但这样的事情可能会起作用：

static readonly ConcurrentDictionary<string, Task<Task>> tasks = new();

//

var newTask = new Task<Task>(() => DoSomethingAsync());

if (this.tasks.TryAdd(taskId, newTask))
{
    newTask.Start();
}

var somethingTask = await this.tasks[taskId];
await somethingTask;

它假定任务永远不会被删除。

基本上有一个外部任务，只有比赛获胜者才能开始。如果正在完成的工作是异步的，例如创建文件或网络资源，则其中需要有另一个 Task。

这个很好。