线程安全的任务队列等待

Question

我想做一件简单的事情（假设ContinueWith是thread-safe）：

readonly Task _task = Task.CompletedTask;

// thread A: conditionally prolong the task
if(condition)
    _task.ContinueWith(o => ...);

// thread B: await for everything
await _task;

问题：在上面的代码中await _task立即返回，忽略是否有inner任务。

_task 可以延长 多个 ContinueWith 的扩展要求，我将如何等待所有这些都完成？

---

当然我可以尝试用旧的线程安全方式来做：

Task _task = Task.CompletedTask;
readonly object _lock = new object();

// thread A
if(condition)
    _lock(_lock)
        _task = _task.ContinueWith(o => ...); // storing the latest inner task

// thread B
lock(_lock)
    await _task;

或者通过将任务存储在线程安全集合中并使用Task.WaitAll。

但我很好奇第一个 sn-p 是否有简单的修复方法？

Answer 1

你的旧版本没问题，除了await同时保持锁；您应该在持有锁的同时将 _task 复制到局部变量中，然后释放锁，然后再将 await

但 ContinueWith 工作流程不是 IMO 实现该逻辑的最佳方式。在 async-await 成为语言的一部分之前，ContinueWith 是在 .NET 4.0 中引入的。

在现代 C# 中，我认为最好这样做：

static async Task runTasks()
{
    if( condition1 )
        await handle1();
    if( condition2 )
        await handle2();
}
readonly Task _task = runTasks();

或者如果您希望它们并行运行：

IEnumerable<Task> parallelTasks()
{
    if( condition1 )
        yield return handle1();
    if( condition2 )
        yield return handle2();
}
readonly Task _task = Task.WhenAll( parallelTasks() );

附：如果条件是动态变化的，对于顺序版本，您应该在第一次等待之前将它们的值复制到局部变量。

更新：所以你是说你的线程 A 延长了该任务以响应一些动态发生的事件？如果是，您的旧方法还可以，简单且有效，只需修复 lock(){ await ... } 并发错误即可。

理论上，更简洁的方法可能类似于反应器模式，但不幸的是它要复杂得多。这是我的open source example 有点类似的东西。您不需要并发限制信号量，也不需要第二个队列，但是您需要公开一个 Task 类型的属性，使用 CompletedTask 对其进行初始化，并在发布第一个时用 post 方法中的TaskCompletionSource<bool> 创建的新任务替换，并在没有更多发布的任务时在 runAsyncProcessor 方法中完成该任务。

Answer 2

在您的原始代码中，await _task; 立即返回，因为 task 已完成。你需要做的

_task = _task.ContinueWith(...);

您不必使用锁。您正在同一个线程中工作。在代码的“锁定”版本中，您确实使用了_task = _task.ContinueWith(...)，这就是它起作用的原因，而不是因为锁定。

---

编辑：刚刚看到您想在一个线程中执行 _task = _task.ContinueWith(...); 和 _task = _task.ContinueWith(...)。

这在设计上是相当糟糕的。你永远不应该将线程和任务结合起来。您应该选择仅任务解决方案或仅线程解决方案。

如果您选择仅任务解决方案（推荐），只需创建一个良好的工作任务序列，然后在每个任务上根据前一个任务的结果决定如何进行。

如果您选择纯线程解决方案，您可能希望使用ManualResetEvent 句柄来同步您的任务。