c# Threadpool - 限制线程数

c# Threadpool - limit number of threads

提问by Bryan

I am developing a console app.

我正在开发一个控制台应用程序。

I want to use a Threadpool to perform web downloads. Here is some fake code.

我想使用线程池来执行网络下载。这是一些假代码。

 for (int loop=0; loop< 100; loop++)
 {
     ThreadPool.QueueUserWorkItem(new WaitCallback(GetPage), pageList[loop]);
 }


snip

private static void GetPage(object o)
{
    //get the page
}

How do I prevent my code from starting more than two (or ten, or whatever) simultaneous threads?

如何防止我的代码启动两个以上（或十个，或其他）并发线程？

I have tried

我试过了

    ThreadPool.SetMaxThreads(1, 0);
    ThreadPool.SetMinThreads(1, 0);

But they seem to have no impact.

但它们似乎没有任何影响。

采纳答案by Brian Gideon

I would use Parallel.Forand set MaxDegreeOfParallelismaccordingly.

我会相应地使用Parallel.For和设置MaxDegreeOfParallelism。

Parallel.For(0, 1000, new ParallelOptions { MaxDegreeOfParallelism = 10 },
  i =>
  {
    GetPage(pageList[i]);
  });

回答by dknaack

Description

描述

You can do this using the ThreadPool.SetMaxThreadsmethod.

您可以使用该ThreadPool.SetMaxThreads方法执行此操作。

But there are some problems using the ThreadPool for WebRequest. Read, for example, this (Bug in ThreadPool or HttpWebRequest?)

但是将 ThreadPool 用于 WebRequest 存在一些问题。阅读，例如，this (Bug in ThreadPool or HttpWebRequest?)

Sample

样本

ThreadPool.SetMaxThreads(2,2);

Edit

编辑

Personally i would use AsParallel from Linq, for this.

就个人而言，我会为此使用 AsParallel Linq。

More Information

更多信息

• MSDN - ThreadPool.SetMaxThreads Method
• MSDN - ParallelEnumerable.AsParallel Method

• MSDN - ThreadPool.SetMaxThreads 方法
• MSDN - ParallelEnumerable.AsParallel 方法

回答by Chris Gessler

Personally, I would use SmartThreadPooland leave the ThreadPool alone. However, this is probably what you want: C# thread pool limiting threads

就个人而言，我会使用SmartThreadPool而不要管 ThreadPool。但是，这可能就是您想要的：C# 线程池限制线程

Included code from link (please give the original author credit, not me)

链接中包含的代码（请注明原作者，而不是我）

System.Threading.Semaphore S = new System.Threading.Semaphore(3, 3);  
try 
{     
  // wait your turn (decrement)     
  S.WaitOne();     
  // do your thing 
}  
finally 
{     
  // release so others can go (increment)     
  S.Release(); 
} 

回答by blockloop

Look at the parameters of ThreadPool.SetMaxThreads. The first parameter is the amount of worker threads and the second parameter is the amount of async threads, which is which one you're talking about.

查看ThreadPool.SetMaxThreads的参数。第一个参数是工作线程的数量，第二个参数是异步线程的数量，也就是你说的那个。

Further down the documentation, it says:

在文档的进一步下方，它说：

You cannot set the number of worker threads or the number of I/O completion threads to a number smaller than the number of processors in the computer.

您不能将工作线程数或 I/O 完成线程数设置为小于计算机中处理器数的数字。

It sounds like you're trying to use the ThreadPool for something it's not intended to be used for. If you want to limit the amount of downloads create a class that manages this for you, because the ThreadPool isn't necessarily the complete solution to your problem.

听起来您正在尝试将 ThreadPool 用于它不打算用于的东西。如果您想限制下载量，请创建一个为您管理此内容的类，因为 ThreadPool 不一定是您问题的完整解决方案。

I'd suggest a class that starts two threads in the ThreadPool and waits for the callback. When it receives a callback for the completion of one of the threads queue a new one.

我建议在 ThreadPool 中启动两个线程并等待回调的类。当它收到一个线程完成的回调时，一个新的线程排队。

回答by mistika

If you are tighten to .Net 2.0 you can use the following technique:

如果您使用 .Net 2.0，您可以使用以下技术：

knowing the fact that whey you enqueue a task to the ThreadPoolit will create a new thread (of course if there're no free ones), you'll wait before doing this until there's a free thread. For this purpose the BlockingCounterclass is used (described below) that once the limit is reached will wait to increment until someone (another thread) decrements it. Then it's entered "closed" state indicating that no new Increments will be done and waits for completion.

知道将任务排入队列的事实ThreadPool将创建一个新线程（当然如果没有空闲线程），您将在执行此操作之前等待直到有空闲线程。为此目的，使用BlockingCounter该类（如下所述），一旦达到限制，将等待递增，直到有人（另一个线程）递减它。然后它进入“关闭”状态，表示不会完成新的增量并等待完成。

Below is the sample that shows a maximum of 4 tasks with the total number of 10.

下面的示例显示了最多 4 个任务，总数为 10。

class Program
{

    static int s_numCurrentThreads = 0;

    static Random s_rnd = new Random();

    static void Main(string[] args)
    {

        int maxParallelTasks = 4;
        int totalTasks = 10;

        using (BlockingCounter blockingCounter = new BlockingCounter(maxParallelTasks))
        {
            for (int i = 1; i <= totalTasks; i++)
            {

                Console.WriteLine("Submitting task {0}", i);
                blockingCounter.WaitableIncrement();
                if (!ThreadPool.QueueUserWorkItem((obj) =>
                                                      {
                                                          try
                                                          {
                                                              ThreadProc(obj);
                                                          }
                                                          catch (Exception ex)
                                                          {
                                                              Console.Error.WriteLine("Task {0} failed: {1}", obj, ex.Message);
                                                          }
                                                          finally
                                                          {
                                                              // Exceptions are possible here too, 
                                                              // but proper error handling is not the goal of this sample
                                                              blockingCounter.WaitableDecrement();
                                                          }
                                                      }, i))
                {
                    blockingCounter.WaitableDecrement();
                    Console.Error.WriteLine("Failed to submit task {0} for execution.", i);
                }
            }

            Console.WriteLine("Waiting for copmletion...");
            blockingCounter.CloseAndWait(30000);
        }

        Console.WriteLine("Work done!");
        Console.ReadKey();

    }

    static void ThreadProc (object obj)
    {
        int taskNumber = (int) obj;
        int numThreads = Interlocked.Increment(ref s_numCurrentThreads);

        Console.WriteLine("Task {0} started. Total: {1}", taskNumber, numThreads);
        int sleepTime = s_rnd.Next(0, 5);
        Thread.Sleep(sleepTime * 1000);
        Console.WriteLine("Task {0} finished.", taskNumber);

        Interlocked.Decrement(ref s_numCurrentThreads);
    }

It uses the BlockingCounter class that is based on the Marc Gravell's SizeQueue posted here, but without a counter instead of a queue. When you end queueing new threads call Close() method and then wait for it to finish.

它使用是基于马克Gravell的SizeQueue的BlockingCounter类张贴在这里，但没有一个计数器，而不是一个队列。当您结束排队新线程时，调用 Close() 方法，然后等待它完成。

public class BlockingCounter : IDisposable
{
    private int m_Count;
    private object m_counterLock = new object();

    private bool m_isClosed = false;
    private volatile bool m_isDisposed = false;

    private int m_MaxSize = 0;

    private ManualResetEvent m_Finished = new ManualResetEvent(false);

    public BlockingCounter(int maxSize = 0)
    {
        if (maxSize < 0)
            throw new ArgumentOutOfRangeException("maxSize");
        m_MaxSize = maxSize;
    }


    public void WaitableIncrement(int timeoutMs = Timeout.Infinite)
    {
        lock (m_counterLock)
        {
            while (m_MaxSize > 0 && m_Count >= m_MaxSize)
            {
                CheckClosedOrDisposed();
                if (!Monitor.Wait(m_counterLock, timeoutMs))
                    throw new TimeoutException("Failed to wait for counter to decrement.");
            }

            CheckClosedOrDisposed();
            m_Count++;

            if (m_Count == 1)
            {
                Monitor.PulseAll(m_counterLock);
            }

        }
    }

    public void WaitableDecrement(int timeoutMs = Timeout.Infinite)
    {
        lock (m_counterLock)
        {
            try
            {
                while (m_Count == 0)
                {
                    CheckClosedOrDisposed();
                    if (!Monitor.Wait(m_counterLock, timeoutMs))
                        throw new TimeoutException("Failed to wait for counter to increment.");
                }

                CheckDisposed();

                m_Count--;

                if (m_MaxSize == 0 || m_Count == m_MaxSize - 1)
                    Monitor.PulseAll(m_counterLock);
            }
            finally
            {
                if (m_isClosed && m_Count == 0)
                    m_Finished.Set();
            }
        }
    }

    void CheckClosedOrDisposed()
    {
        if (m_isClosed)
            throw new Exception("The counter is closed");
        CheckDisposed();
    }

    void CheckDisposed()
    {
        if (m_isDisposed)
            throw new ObjectDisposedException("The counter has been disposed.");
    }

    public void Close()
    {
        lock (m_counterLock)
        {
            CheckDisposed();
            m_isClosed = true;
            Monitor.PulseAll(m_counterLock);
        }
    }

    public bool WaitForFinish(int timeoutMs = Timeout.Infinite)
    {
        CheckDisposed();
        lock (m_counterLock)
        { 
             if (m_Count == 0)
                 return true;
        }
        return m_Finished.WaitOne(timeoutMs);
    }

    public void CloseAndWait (int timeoutMs = Timeout.Infinite)
    {
        Close();
        WaitForFinish(timeoutMs);
    }

    public void Dispose()
    {
        if (!m_isDisposed)
        {
            m_isDisposed = true;
            lock (m_counterLock)
            {
                // Wake up all waiting threads, so that they know the object 
                // is disposed and there's nothing to wait anymore
                Monitor.PulseAll(m_counterLock);
            }
            m_Finished.Close();
        }
    }
}

The result will be like that:

结果将是这样的：

Submitting task 1
Submitting task 2
Submitting task 3
Submitting task 4
Submitting task 5
Task 1 started. Total: 1
Task 1 finished.
Task 3 started. Total: 1
Submitting task 6
Task 2 started. Total: 2
Task 3 finished.
Task 6 started. Total: 4
Task 5 started. Total: 3
Task 4 started. Total: 4
Submitting task 7
Task 4 finished.
Submitting task 8
Task 7 started. Total: 4
Task 5 finished.
Submitting task 9
Task 7 finished.
Task 8 started. Total: 4
Task 9 started. Total: 4
Submitting task 10
Task 2 finished.
Waiting for copmletion...
Task 10 started. Total: 4
Task 10 finished.
Task 6 finished.
Task 8 finished.
Task 9 finished.
Work done!

回答by Diego Troiti?o

Just inverse that code from:

只需反转该代码：

ThreadPool.SetMaxThreads(1, 0);
ThreadPool.SetMinThreads(1, 0);

To:

到：

ThreadPool.SetMinThreads(1, 0);
ThreadPool.SetMaxThreads(1, 0);

You can't set the MaxThread smaller than MinThread

您不能将 MaxThread 设置为小于 MinThread