如何在Worker中設置多個線程並向其發送消息？

Question

我在我的系統中有很多「長期運行但未綁定CPU」的「作業」。我想設置一個Worker角色來處理這些角色，但它們具有足夠的可擴展性，以至於一個Worker角色可以輕鬆地擁有10-20個線程同時處理「作業」。

這裏有一些問題建議使用TPL，我確實有一些有限的經驗。然而，我不明白的是如何管理這些線程，以便有最大數量的線程，以及在釋放線程時如何分配它們。

稍微複雜一點是我想用Ninject來創建每個「工作」所需的服務。

下面是我形象在我的頭上的工作：

while (true) 
{ 
    // Don't go unless we have a free slot (how do I implement this?!) 
    if (FreeThreadExists) 
    { 
     // Get the next message 
     CloudQueueMessage ThisMessage = Queue.GetMessage(TimeSpan.FromMinutes(3)); 

     // Get the new job and inject services 
     Job MyJob = Kernel.Get<Job>(); 

     // Start this job 
     // Will I need to keep ahold of this Task? 
     // And how do I know when it's done so that FreeThreadExists changes? 
     Task.Factory.StartNew(() => MyJob.Run(ThisMessage)); 
    } 
    else 
    {  
     // Sleep to prevent choking 
     Thread.Sleep(500); 
    } 
} 

然後在該線程刪除完成的消息。基本上我試圖將一個Worker分成20個「實例」，而不會失去太多的Azure功能（特別是我希望隊列消息超時/重試功能）。

我對.NET線程並不熟悉，對此有什麼最好的方法呢？

編輯：哇，我完全忘了補充一個重要的觀點：這需要跨越多個工作人員。所以，10個Worker角色，每個都有10個線程，消息被UI前端排隊，然後由第一個Worker以一個空閒線程出列並運行。

Answer 1

試試這個僞代碼：

while (true) 
{ 
    int maxThreadsPerWorkerRole = 3;//assuming each worker role can handle 3 jobs simultaneously 
    var messages = Queue.GetMessages(3);//Get 3 messages from the queue 
    if (messages != null && messages.Count > 0)//Ensuring there is some work which needs to be done 
    { 
     var myTasks = new List<Task>(); 
     for (int i=0; i<messages.Count; i++) 
     { 
      Job MyJob = Kernel.Get<Job>();//Get the job 
      var task = Task.Factory.StartNew(() => MyJob.Run(messages[i])); 
      myTasks.Add(task); 
     } 
     Task.WaitAll(myTasks.ToArray());//Wait for all tasks to complete. 
     for (int i=0; i<messages.Count; i++) 
     { 
      //Write code to delete the message. 
     } 
     //Check if the queue is empty or not. If the queue is not empty, then repeat this loop 
     //Otherwise simply exit this loop. 
     if (Queue.RetrieveApproximateMessageCount() == 0) 
     { 
      break; 
     } 
    } 
} 

希望這有助於。

Answer 2

我通常也會涉及TPL，就像您在代碼中提到的那樣。作爲Gaurav方法的替代方法，請參閱下面的代碼。下面的僞代碼使用Parallel.For來控制線程的創建。每個線程啓動並運行一個無限循環;如果沒有工作，確保你睡了一下。

// Start 10 threads 
Parallel.For(0, 10, (i) => 
{ 
    while (true) 
    { 
    // Get message from queue 
    var msg = Queue.GetMessage(); 
    if (msg != null) 
    { 
     // Do some work here... 
     StartSomeJob(); 

     // Then when you are done, delete the message 
     Queue.DeleteMessage(msg); 
    } 
    // Wait 1 second before fetching next work item from queue 
    System.Threading.Thread.Sleep(1000); 
    } 
});