多线程：限制并发线程数

Question

我需要开发一个使用多线程的应用程序。

基本上，我有一个包含大约 200k 行的 DataTable。 我需要从每一行中取出一个字段，将其与网页进行比较， 然后将其从数据表中删除。

问题是，为这些页面提供服务的服务器对并发请求有限制。 所以我最多可以同时请求 3 页。

我想通过使用线程池来做到这一点， 我什至设法构建了一个简单的应用程序来做到这一点（锁定数据表） 但我无法限制并发线程（即使使用 SetMaxThreads ），它似乎只是忽略了限制。

有人有现成的东西可以做类似的事情吗？ 我很想看看。

我尝试过使用信号量，但遇到了问题：

        static SemaphoreSlim _sem = new SemaphoreSlim(3);    // Capacity of 3
    static List<string> records = new List<string>();

    static void Main()
    {
        records.Add("aaa");
        records.Add("bbb");
        records.Add("ccc");
        records.Add("ddd");
        records.Add("eee");
        records.Add("fff");
        records.Add("ggg");
        records.Add("iii");
        records.Add("jjj");

        for (int i = 0; i < records.Count; i++ )
        {
            new Thread(ThreadJob).Start(records[i]);
        }

        Console.WriteLine(records.Count);
        Console.ReadLine();
    }

    static void ThreadJob(object id)
    {
        Console.WriteLine(id + " wants to enter");
        _sem.Wait();
        Console.WriteLine(id + " is in!");           // Only three threads
        //Thread.Sleep(1000 * (int)id);               // can be here at
        Console.WriteLine(id + " is leaving");       // a time.

        lock (records)
        {
            records.Remove((string)id);
        }

        _sem.Release();
    }

运行得很好，唯一的问题是

Console.WriteLine(records.count);

返回不同的结果。 即使我知道这种情况发生，因为并非所有线程都已完成（在删除所有记录之前，我正在调用records.count），我找不到如何等待所有线程完成。

Answer 1

要等待多个线程完成，您可以使用多个 EventWaitHandle，然后调用 WaitHandle.WaitAll 阻塞主线程，直到所有事件都发出信号：

// we need to keep a list of synchronization events
var finishEvents = new List<EventWaitHandle>();

for (int i = 0; i < records.Count; i++ )
{
    // for each job, create an event and add it to the list
    var signal = new EventWaitHandle(false, EventResetMode.ManualReset);
    finishEvents.Add(signal);

    // we need to catch the id in a separate variable
    // for the closure to work as expected
    var id = records[i];

    var thread = new Thread(() =>
        {
            // do the job
            ThreadJob(id);

            // signal the main thread
            signal.Set();
        });
}

WaitHandle.WaitAll(finishEvents.ToArray());

由于大多数这些线程在大多数情况下都会挂起，因此最好在这种情况下使用 ThreadPool，因此您可以将 new Thread 替换为：

    ThreadPool.QueueUserWorkItem(s =>
    {
        ThreadJob(id);
        signal.Set();
    });

当您完成事件时，不要忘记处理它们：

foreach (var evt in finishEvents)
{
    evt.Dispose();
}

[编辑]

为了将所有内容放在一个地方，您的示例代码应如下所示：（

static Semaphore _sem = new Semaphore(3, 3);    // Capacity of 3
static List<string> _records = new List<string>(new string[] { "aaa", "bbb", "ccc", "ddd", "eee", "fff", "ggg", "hhh" });

static void Main()
{
    var finishEvents = new List<EventWaitHandle>();

    for (int i = 0; i < _records.Count; i++)
    {
        var signal = new EventWaitHandle(false, EventResetMode.ManualReset);
        finishEvents.Add(signal);

        var id = _records[i];
        var t = new Thread(() =>
        {
            ThreadJob(id);
            signal.Set();
        });

        t.Start();
    }

    WaitHandle.WaitAll(finishEvents.ToArray());

    Console.WriteLine(_records.Count);
    Console.ReadLine();
}

static void ThreadJob(object id)
{
    Console.WriteLine(id + " wants to enter");
    _sem.WaitOne();

    Console.WriteLine(id + " is in!");
    Thread.Sleep(1000);
    Console.WriteLine(id + " is leaving");

    lock (_records)
    {
        _records.Remove((string)id);
    }

    _sem.Release();
}

请注意，我使用了 Semaphore 而不是 SemaphoreSlim，因为我不这台机器上没有 .NET 4，我想在更新答案之前测试代码）

Answer 2

为什么不使用并行扩展 - 这会让事情变得更容易。

不管怎样，你可能想看看信号量之类的东西。一两个月前我写了一篇关于这个主题的博客文章，您可能会觉得有用：https://colinmackay.scot/2011/03/30/using-semaphores-to-restrict-access-to-resources/

Answer 3

你可以使用
Semaphore（如果您使用的是 .net 3.5）

或

.net 4.0 中的 SemaphoreSlim代码>

Answer 4

首先，应该
Console.WriteLine(id + "要离开");
不是晚一点，在锁定之后、释放信号量之前吗？

至于实际等待所有线程完成，从长远来看，Groo 的答案看起来更好、更稳健，但作为此特定代码段的更快/更简单的解决方案，我认为您也可以只调用 .按顺序对您想要等待的所有线程进行 Join() 。

static List<Thread> ThreadList = new List<Thread>(); // To keep track of them

然后在启动线程时，将当前的新线程行替换为：

ThreadList.Add(new Thread(ThreadJob).Start(records[i]));

然后在 Console.WriteLine 之前：

foreach( Thread t in ThreadList )
{
    t.Join();
}

如果任何线程没有终止，并且如果您想知道哪个线程还没有终止，则这将锁定还没完，这个方法不行。