コンソールアプリを開発しています。
Threadpool を使用して Web ダウンロードを実行したいと考えています。ここにいくつかの偽のコードがあります。
for (int loop=0; loop< 100; loop++)
{
ThreadPool.QueueUserWorkItem(new WaitCallback(GetPage), pageList[loop]);
}
snip
private static void GetPage(object o)
{
//get the page
}
コードで 2 つ (または 10 など) を超える同時スレッドを開始しないようにするにはどうすればよいですか?
私が試してみました
ThreadPool.SetMaxThreads(1, 0);
ThreadPool.SetMinThreads(1, 0);
しかし、それらは影響を与えていないようです。
それに応じて使用Parallel.Forおよび設定しMaxDegreeOfParallelismます。
Parallel.For(0, 1000, new ParallelOptions { MaxDegreeOfParallelism = 10 },
i =>
{
GetPage(pageList[i]);
});
個人的には、SmartThreadPoolを使用し、ThreadPool はそのままにしておきます。ただし、これはおそらくあなたが望むものです: C# スレッド プール制限スレッド
リンクから含まれるコード (私ではなく、元の作者のクレジットを与えてください)
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();
}
ThreadPool.SetMaxThreadsメソッドを使用してこれを行うことができます。
ただし、WebRequest に ThreadPool を使用すると、いくつかの問題があります。たとえば、これを読んでください（ThreadPoolまたはHttpWebRequestのバグ？）
ThreadPool.SetMaxThreads(2,2);
個人的には、これには AsParallel from を使用Linqします。
ThreadPool.SetMaxThreads のパラメーターを見てください。最初のパラメーターはワーカー スレッドの量で、2 番目のパラメーターは非同期スレッドの量です。
ドキュメントのさらに下には、次のように書かれています。
ワーカー スレッドの数または I/O 完了スレッドの数を、コンピューターのプロセッサの数よりも小さい数に設定することはできません。
ThreadPool を使用することを意図していないものに使用しようとしているようです。ダウンロードの量を制限したい場合は、これを管理するクラスを作成してください。ThreadPool が問題の完全な解決策であるとは限らないためです。
ThreadPool で 2 つのスレッドを開始し、コールバックを待機するクラスをお勧めします。スレッドの 1 つの完了のコールバックを受信すると、新しいスレッドが待ち行列に入れられます。
.Net 2.0 に固執している場合は、次の手法を使用できます。
タスクをキューに入れると新しいスレッドが作成されるという事実を知っているのでThreadPool(もちろん空きスレッドがない場合)、空きスレッドができるまでこれを行う前に待機します。この目的のためにBlockingCounter、制限に達すると誰か (別のスレッド) がデクリメントするまでインクリメントを待機するクラスが使用されます (後述)。次に、「クローズ」状態に入り、新しいインクリメントが実行されず、完了を待機することを示します。
以下は、合計数が 10 の最大 4 つのタスクを示すサンプルです。
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);
}
ここに投稿された Marc 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();
}
}
}
結果は次のようになります。
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!