创建一个可等待的冷任务

Question

我有一个异步方法，完成后我希望运行另一个方法。如果我简单地调用该方法并添加 .ContinueWith()

，这很好用

但是，我有一个新要求，即只有在我能够将任务添加到并发字典时才开始任务。

我希望构建任务，尝试添加它然后启动任务

但是，Task.Start() 似乎立即完成了导致继续操作运行的任务，并且任何等待......而不是等待。

谁能解释为什么会发生这种情况以及实现我的目标的正确方法？

namespace UnitTestProject2
{
    [TestClass]
    public class taskProblem
    {
        [TestMethod]
        public void Test()
        {
            CancellationTokenSource cancel = new CancellationTokenSource();
            ConcurrentDictionary<Guid, Task> tasks = new ConcurrentDictionary<Guid,Task>();
            Guid id = Guid.NewGuid();
            Task t = new Task(async () => await Get(), cancel.Token);
            t.ContinueWith(Complete);
            if (tasks.TryAdd(id, t))
            {
                t.Start();
            }
            else
            {
                //another thread is stopping stuff dont start new tasks
            }

            t.Wait(); //expected to wait for the get function to complete
            Console.WriteLine("end test");
        }

        public async Task Get()
        {
            Console.WriteLine("start task");
            await Task.Delay(10000);
            Console.WriteLine("end task");
        }

        public void Complete(Task t)
        {
            Console.WriteLine("Complete");
        }
    }
}

输出：

start task
end test
Complete

预期输出：

start task
end task
Complete
end test

更新：似乎没有办法创建一个不会立即在 Task.Start 上立即启动或完成的新任务？

Answer 1

您的委托是异步无效的。 异步 void 方法是一劳永逸的方法。

参见模式和反模式总结的第一点：http://rarcher.azurewebsites.net/Post/PostContent/31

也许你可以这样做：

[TestFixture]
public class FIXTURENAMETests {
  [Test]
  public async Task NAME() {
    var tcs = new TaskCompletionSource<bool>();
    Task t = LongRunningStuff(tcs);

    if (CanInsertInDictionary(t)) {
      tcs.SetResult(true);
    } else {
      tcs.SetException(new Exception());
    }

    Trace.WriteLine("waiting for end");

    try {
      await t;
    }
    catch (Exception exception) {
      Trace.WriteLine(exception);
    }

    Trace.WriteLine("end all");
  }

  private bool CanInsertInDictionary(Task task) {
    return true;
  }

  private async Task LongRunningStuff(TaskCompletionSource<bool> tcs) {
    Trace.WriteLine("start");
    try {
      await tcs.Task;
    }
    catch (Exception) {
      return;
    }
    Trace.WriteLine("do long running stuff");
    await Task.Delay(10000);
    Trace.WriteLine("end");
  }
}

Answer 2

但是，我有一个新要求，即只有在我能够将任务添加到并发字典时才开始任务。

这是上述问题的一种可能的解决方案：

• 将Task<T> 包裹在Lazy<> 中
• 通过GetOrAdd将惰性任务添加到ConcurrentDictionary
• 使用GetOrAdd 的输出来等待任务

这样，当ConcurrentDictionary 正在执行它的内部工作以处理冲突等时，任务将不会启动，并且所有后续await 将使用与字典中给定键关联的同一个任务对象.

总之，在多个线程向字典添加任务的情况下，下面的代码只允许对字典中的给定键执行一次任务主体。

代码如下：

[TestMethod]
public void ConcurrentMapLazyTask() {

  Func<Task> Get = async () => {
    Trace.WriteLine("Start task.");
    await Task.Delay(200);
    Trace.WriteLine("End task.");
  };

  Action<Task> Complete = (t) => {
    Trace.WriteLine("Complete.");
  };

  var mp = new ConcurrentDictionary<string, Lazy<Task>>();

  Func<string, Lazy<Task>> valueFactory = (sKey) => {
    var s = Guid.NewGuid().ToString();
    Trace.WriteLine(string.Format("valueFactory called => {0}", s));
    return new Lazy<Task>(() => {
      Trace.WriteLine(string.Format("LazyTask factory called for {0}", s));
      var t = Task.Run(async () => {
        Trace.WriteLine(string.Format("Task.Run executed for {0}", s));
        await Get();
      });
      return Task.WhenAll(t, t.ContinueWith(Complete));
    });
  };

  Func<Task> TestAsync = async () => {
    var lazyTask = mp.GetOrAdd("test", valueFactory);
    await lazyTask.Value;
  };

  Action TestSync = () => {
    TestAsync().Wait();
    Trace.WriteLine("End test.");
  };
  Action TestSyncSlow = () => {
    Thread.Sleep(50);
    TestAsync().Wait();
    Trace.WriteLine("End slow test.");
  };
  //Parallel.Invoke(TestSync);
  Parallel.Invoke(TestSync, TestSyncSlow, TestSync, TestSync);
}

这是输出：

valueFactory called => e35b0e9e-3326-41bb-b3b9-eb4e6dc37391
valueFactory called => e4929d93-4c77-4fbf-8e90-e8d6e4d7b009
LazyTask factory called for e35b0e9e-3326-41bb-b3b9-eb4e6dc37391
valueFactory called => 8863fb7f-1309-4c27-b805-71467022ac74
Task.Run executed for e35b0e9e-3326-41bb-b3b9-eb4e6dc37391
Start task.
End task.
Complete.
End test.
End test.
End slow test.
End test.

• Task.Run 只执行一次
• Lazy factory 只被调用过一次
• 在GetOrAdd 中使用valueFactory 参数可以减少创建的惰性对象的数量（3 个惰性包装器由4 次并发调用GetOrAdd 产生）

Answer 3

首先，ContinueWith 将返回一个新的Task，您想等到Complete 方法完成，但您正在等待第一个任务t。

所以，要在end test 之前输出Complete，你必须等待第二个任务：

Task t = new Task(async () => await Get(), cancel.Token);

// NOTE: t2 is a new Task returned from ContinueWith
Task t2 = t.ContinueWith(Complete);

if (tasks.TryAdd(id, t))
{
    t.Start();
}
else
{
}

// NOTE: Waiting on t2, NOT t
t2.Wait();

Console.WriteLine("end test");

现在输出将是：

start task 
Complete
end test
end task

好的，这仍然不是预期的输出。 end task 应该在Complete 之前打印。这是因为您的异步操作不可等待：How to await on async delegate

不知道我是否正确理解了您的要求。如果是我，我可能会这样做：

添加一个新的支持类：

public class TaskEntry
{
    public Task Task { get; set; }
}

然后将代码更改为：

Guid id = Guid.NewGuid();

Task task = null;
var entry = new TaskEntry();
if (tasks.TryAdd(id, entry))
{
    entry.Task = Get();

    // Notice this line of code:
    task = entry.Task.ContinueWith(Complete);
}

if (task != null)
{
    task.Wait();
}

Console.WriteLine("end test");

这里我假设TaskEntry 不会被其他线程修改。