等待所有线程完成

Question

我在这段代码中遇到了一些关于线程的问题。

我想要同时运行很多任务，并在所有任务都执行完毕后继续。

while (true)
{
    //   Run tasks together:
    foreach (object T in objectsList)
    {
        if (T.something>0)
            var task = Task.Factory.StartNew(() => T.RunObject());
            task.ContinueWith(delegate { ChangeObject(T, 1); }, TaskContinuationOptions.NotOnFaulted);
    }
    //   <-- Here I want to wait for all the task to be finish.
    //   I know its task.Wait() but how to waitAll()?
    System.Threading.Thread.Sleep(this.GetNextTime()); 
    var RefreshObjects = new Task(loadObjectsList); RefreshObjects .Start(); RefreshObjects.Wait();
}

我不知道 objectsList 中会有多少对象，也不知道 T.something 是否大于 0。因此我不能简单地使用以下方法：

Task[] Tasks = new Task[objectsList.count()]
for (int T=0; T<objectsList.count(); ++T)
{
    if (objectsList[T].something>0)
        var task = Task.Factory.StartNew(() => objectsList[T].RunObject());
        task.ContinueWith(delegate { ChangeObject(objectsList[T], 1); }, ...);
}
Task.WaitAll(Tasks);

因为在objectsList[T].something!>0时，Tasks将包含空值......谢谢任何建议！

- Yaron

Answer 1

只需切换条件并创建一个任务列表，仅为符合您标准的对象创建。

var tasks = objectsList
            .Where(x => x.Something() > 0)
            .Select(x => {
                            var task = Task.Factory.StartNew(() => x.RunObject());
                            task.ContinueWith(t => ChangeObject(....));
                            return task;
                         })
            .ToArray();

    Task.WaitAll(tasks);

您的代码示例只等待RunObject()完成！如果您希望如此，请跳过我的其余答案。如果您希望等待连续完成，也可以使用以下内容

var tasks = objectsList
            .Where(x => x.Something() > 0)
            .Select(x => Task.Factory.StartNew(() => x.RunObject()).ContinueWith(t => ChangeObject(....)))
            .ToArray();

    Task.WaitAll(tasks);

因为ContinueWith生成了一个新的任务。

- Zebi

---

你可以通过在任务上调用ContinueWith而不将其分配给变量来轻松避免使select语句成为委托。 - Servy

取决于他的需求。如果他想等待连续任务完成，那么他可以选择等待。如果他只想等待初始任务完成，那么他必须使用该语句。 - Zebi

非常有帮助！谢谢。 - Yaron

Answer 2

如果objectsList实现了IEnumerable（就像数组一样），（并且列表中的对象少于64个），则可以使用以下方法：

public delegate void SyncDelegatesInParallelDelegate<in T>(T item);
public static class ParallelLinqExtensions
{
    public static void SyncDelegatesInParallel<T>(
        this IEnumerable<T> list, 
        SyncDelegatesInParallelDelegate<T> action)
    {
        var foundCriticalException = false;
        Exception exception = null;
        var waitHndls = new List<WaitHandle>();

        foreach (var item in list)
        {
            // Temp copy of session for modified closure
            var localItem = item;
            var txEvnt = new ManualResetEvent(false);

            // Temp copy of session for closure
            ThreadPool.QueueUserWorkItem(
                 depTx =>
                 {
                     try { if (!foundCriticalException) action(localItem); }
                     catch (Exception gX) 
                     { exception = gX; foundCriticalException = true; }
                     finally { txEvnt.Set(); }
                 }, null);
            waitHndls.Add(txEvnt);
        }
        if (waitHndls.Count > 0) WaitHandle.WaitAll(waitHndls.ToArray());
        if (exception != null) throw exception;
    }
}

你会这样调用它。

 objectsList.SyncDelegatesInParallel(delegate { ChangeObject(T, 1);});

- Charles Bretana

---

如果objectsList中的对象超过64个，那么这将无法工作，因为等待句柄的数量是有限制的。 - Charles Bretana