如何停止一个线程直到n个线程完成其工作

.NET 4.0将包括 System.Threading.Barrier 使多个线程之间的同步更容易的类。可以找到一篇包含一些很好示例代码的博客文章 here .

Similar functionality can be achieved 在.NET 3.0+中使用多个waithandles,如中所示 this example 在MSDN上。

MSDN示例的简要摘要:

const int numberOfWorkers = 5;

static void Main()
{
    var handles = new ManualResetEvent[numberOfWorkers];

    for (int i = 0; i < numberOfWorkers; i++)
    {
        handles[i] = new ManualResetEvent(false);
        ThreadPool.QueueUserWorkItem(o => worker.Work(), null);
    }

    // Wait for all workers to finish before continuing
    WaitHandle.WaitAll(handles);
    /* continue execution... */
}

做一些类似垃圾收集的事情。您将编写一个threadmanager,它计算正在运行的线程数。当主线程启动新的工作线程时,threadmanager将增加其工作线程数。当一个工作进程结束时,它将通知threadmanager,后者将减少线程数。当它没有工作线程时,threadmanager将唤醒主线程。

看起来像 WaitHandle.WaitAll 应该解决这个问题。

您的主线程将需要保留对工作线程等待句柄的引用。当需要同步时,将这些句柄传递到上面的方法中。工作线程在其代码中的适当点发出信号。

如果工作线程循环或需要多次“脉冲”,则可以使用 AutoResetEvents ,像这样:

public void WorkerMethod() {         
     DoFirstThing();
     this.autoResetEvent.Set();
     DoSecondThing();
     this.autoResetEvent.Set();
     // etc.
}

如果没有(如果主线程只是需要知道工作线程已超过某个阈值), ManualResetEvents 就好了。

在使用waitall(来自msdn)时需要注意一些事情 WaitAll 文件):

在某些实现上,如果超过 传递64个句柄,a 引发NotSupportedException。如果 数组包含重复项, 呼叫失败 DuplicateWaitObjectException。

但是,很少有进程能够真正利用64个以上的线程,因此这种限制通常并不重要。

这叫做屏障: http://programmingexamples.wikidot.com/java-barrier

噢,但是如果您只需要第一个线程等待其余线程通过某个点,而您希望另一个线程仍继续工作,那么使用一个大小为n的信号量,让所有其他线程都接受它,而第一个线程则等待在它们之后获取它。

Semaphore: http://programmingexamples.wikidot.com/java-semaphore

因为在某些实现中,句柄的数量是有限制的 WaitHandle.WaitAll() 可以…句柄,(参见 msdn-WaitHandle.WaitAll() ,我为此创建了一个实用方法:

    public static void WaitAll(WaitHandle[] handles)
    {
        if (handles == null)
            throw new ArgumentNullException("handles",
                "WaitHandle[] handles was null");
        foreach (WaitHandle wh in handles) wh.WaitOne();
    }

用法是将每个线程的等待句柄添加到数组中,然后在所有线程启动之后调用上面的实用方法(传递数组)。

 List<WaitHandle> waitHndls = new List<WaitHandle>();
 foreach (MyType mTyp in MyTypeCollection)
 {
     ManualResetEvent txEvnt = new ManualResetEvent(false);
     int qryNo1 = ++qryNo;
     ThreadPool.QueueUserWorkItem(
        delegate
           {
              try
              {
                  // Code to execute whatever thread's function is... 
              }
              catch (SomeCustomException iX)
              {
                  // catch code
              }                                         }
              finally {  lock (locker) txEvnt.Set();  }
           });
     waitHndls.Add(txEvnt);
 }
 util.WaitAll(waitHndls.ToArray());

如果您只需要等待线程终止,那么 Thread.Join ?在.NET 4.0中,您可以使用 Task.WaitAll . 如果你需要等到他们完成任务的一部分,那就有点难了。在.NET的当前版本中,请查看 WaitHandle.WaitAll / Threading.ManualResetEvent . 在.NET 4.0中,您可以使用 Threading.Barrier .

好吧,我现在正在做的(利用你的想法),似乎有效的是:

我宣布了一份ManualResetEvent清单:

Private m_waitHandles As List(Of Threading.ManualResetEvent)

进程接受传入的TCP连接,并在每个连接上启动一个线程。所以在新的客户机处理程序中,我添加了以下代码:

Dim waitHandle As Threading.ManualResetEvent
waitHandle = New Threading.ManualResetEvent(True)
SyncLock (DirectCast(m_waitHandles, IList).SyncRoot)
    m_waitHandles.Add(waitHandle)
End SyncLock

''# Do all the work
StoppableMethod()

SyncLock (DirectCast(m_waitHandles, IList).SyncRoot)
    waitHandle = m_waitHandles.Item(Threading.WaitHandle.WaitAny(m_waitHandles.ToArray()))
End SyncLock

waitHandle.Reset()
NonStoppableMethod()
waitHandle.Set()

SyncLock (DirectCast(m_waitHandles, IList).SyncRoot)
    m_waitHandles.Remove(waitHandle)
End SyncLock

最后一件事是修改stop方法,以确保stop操作不会对不可停止方法内的任何线程执行:

SyncLock (DirectCast(m_waitHandles, IList).SyncRoot)
    If m_waitHandles.Count > 0 Then
        Threading.WaitHandle.WaitAll(m_waitHandles.ToArray())
    End If
End SyncLock

我不确定这样做是否正确,因为这是我第一次处理这样的事情。你觉得这样可以吗?这是一个很好的方法?

谢谢大家,伙计们!

尝试使用这个:

int threadsCompleted = 0;
int numberOfThreads = 4;
ManualResetEvent completedEvent = new ManualResetEvent(false);

在每个线程中:

// Do task

if (Interlocked.Increment(threadsCompleted) == numberOfThreads)
    completedEvent.Set();

主线:

completedEvent.WaitOne();

在互联网上,人们尝试使用一系列 EventHandles 和 WaitAll() . 我想出了下面这门课,这门课的资源要轻得多。我试图考虑不同的种族场景,我相信这段代码中没有种族条件。(在减量和检查条件之间存在理论上的竞争 Count ,但据我所知,它不会影响功能,代码仍然可以正常工作。)

要使用这个类,所有需要同步的线程都必须调用 Wait() 方法。他们会封锁到 伯爵 已调用的线程数 WAIT() . 单个实例只能用于同步一次(不能重置)。

internal class ThreadBarrier
{
    private ManualResetEvent BarrierEvent;
    private int Count;

    internal ThreadBarrier(int count)
    {
        BarrierEvent = new ManualResetEvent(false);
        Count = count;
    }

    internal void Wait()
    {
        Interlocked.Decrement(ref Count);
        if (Count > 0)
            BarrierEvent.WaitOne();
        else
            BarrierEvent.Set();
    }
}

使用 Thread.Join (在线程终止前阻塞调用线程,同时继续执行标准COM和发送消息泵送)方法,如示例中所示:

using System;
using System.Threading;

class IsThreadPool
{
    static void Main()
    {
        AutoResetEvent autoEvent = new AutoResetEvent(false);

        Thread regularThread = 
            new Thread(new ThreadStart(ThreadMethod));
        regularThread.Start();
        ThreadPool.QueueUserWorkItem(new WaitCallback(WorkMethod), 
            autoEvent);

        // __________ Wait for foreground thread to end. __________ 
        regularThread.Join();

        // Wait for background thread to end.
        autoEvent.WaitOne();
    }

    static void ThreadMethod()
    {
        Console.WriteLine("ThreadOne, executing ThreadMethod, " +
            "is {0}from the thread pool.", 
            Thread.CurrentThread.IsThreadPoolThread ? "" : "not ");
    }

    static void WorkMethod(object stateInfo)
    {
        Console.WriteLine("ThreadTwo, executing WorkMethod, " +
            "is {0}from the thread pool.", 
            Thread.CurrentThread.IsThreadPoolThread ? "" : "not ");

        // Signal that this thread is finished.
        ((AutoResetEvent)stateInfo).Set();
    }
}