异步/等待任务和WaitHandle

Question

假设我有10个项目（需要通过http协议获取它们），在代码中启动了N个任务来获取数据，每个任务按顺序获取10个项目。我把这些项目放在一个ConcurrentQueue<Item>中。之后，这些项目将逐个在一个线程不安全的方法中进行处理。

async Task<Item> GetItemAsync()
{
    //fetch one item from the internet
}

async Task DoWork()
{
    var tasks = new List<Task>();
    var items = new ConcurrentQueue<Item>();
    var handles = new List<ManualResetEvent>();

    for i 1 -> N
    {
        var handle = new ManualResetEvent(false);
        handles.Add(handle);

        tasks.Add(Task.Factory.StartNew(async delegate
        {
            for j 1 -> 10
            {
                var item = await GetItemAsync();
                items.Enqueue(item);
            }
            handle.Set();
        });
    }

    //begin to process the items when any handle is set
    WaitHandle.WaitAny(handles);

    while(true)
    {
         if (all handles are set && items collection is empty) //***
           break; 
         //in another word: all tasks are really completed

         while(items.TryDequeue(out item))          
         {
              AThreadUnsafeMethod(item);    //process items one by one
         }
    }
}

我不知道在标记为***的语句中可以放什么if条件。我不能在这里使用Task.IsCompleted属性，因为我在任务中使用了await，所以任务很快就完成了。而一个bool[]数组来表示任务是否执行完全看起来真的很丑，因为我认为ManualResetEvent可以完成同样的工作。有人能给我建议吗？

- Cheng Chen

Answer 1

嗯，你可以自己构建这个功能，但我认为使用TPL Dataflow会更加方便。

像这样：

static async Task DoWork()
{
  // By default, ActionBlock uses MaxDegreeOfParallelism == 1,
  //  so AThreadUnsafeMethod is not called in parallel.
  var block = new ActionBlock<Item>(AThreadUnsafeMethod);

  // Start off N tasks, each asynchronously acquiring 10 items.
  // Each item is sent to the block as it is received.
  var tasks = Enumerable.Range(0, N).Select(Task.Run(
      async () =>
      {
        for (int i = 0; i != 10; ++i)
          block.Post(await GetItemAsync());
      })).ToArray();

  // Complete the block when all tasks have completed.
  Task.WhenAll(tasks).ContinueWith(_ => { block.Complete(); });

  // Wait for the block to complete.
  await block.Completion;
}

- Stephen Cleary

---

Microsoft TPL Dataflow现在被称为System.Threading.Tasks.Dataflow。 - Dude Pascalou

Answer 2

您可以使用零超时时间进行WaitOne以检查状态。类似以下代码应该可以工作：

if (handles.All(handle => handle.WaitOne(TimeSpan.Zero)) && !items.Any())
    break;

http://msdn.microsoft.com/en-us/library/cc190477.aspx

- BAF

---

阻塞当前线程，直到当前实例接收到信号。因此，您将有10个阻塞线程等待信号。 - Sebastian

@Sebastian "备注：如果超时为零，则该方法不会阻塞。它测试等待句柄的状态并立即返回。" - jaggedSpire

@jaggedSpire 好的，这是我错过的信息。谢谢，现在这是我检查 WaitHandles 的首选解决方案。 - Sebastian

Answer 3

感谢大家。最后我发现CountDownEvent非常适合这种情况。一般的实现看起来像这样：（供其他人参考）

for i 1 -> N
{
    //start N tasks
    //invoke CountDownEvent.Signal() at the end of each task
}

//see if CountDownEvent.IsSet here

- Cheng Chen