C＃在任何給定時間運行X個任務<T>，同時保持WinForm響應

Question

我有一個利用TPL的C＃WinForms（.NET 4.5.2）應用程序。該工具具有同步功能，該功能被傳遞到任務工廠X次數（具有不同的輸入參數），其中X是在開始該過程之前用戶聲明的數字。任務開始並存儲在List<Task>中。

假設用戶輸入5，我們有這個在async按鈕單擊處理程序：

for (int i = 0; i < X; i++) 
{ 
    var progress = Progress(); // returns a new IProgress<T> 
    var task = Task<int>.Factory.StartNew(() => MyFunction(progress), TaskCreationOptions.LongRunning); 
    TaskList.Add(task); 
} 

每個progress實例更新UI。

現在，只要一個任務完成，我想要啓動一個新的。從本質上講，這個過程應該無限期地運行，任何時候都有X任務運行，除非用戶通過UI取消（我將使用取消令牌）。我嘗試使用以下方法來實現此目的：

while (TaskList.Count > 0) 
{ 
    var completed = await Task.WhenAny(TaskList.ToArray());         

    if (completed.Exception == null) 
    { 
     // report success 
    } 
    else 
    { 
     // flatten AggregateException, print out, etc 
    } 
    // update some labels/textboxes in the UI, and then: 
    TaskList.Remove(completed); 
    var task = Task<int>.Factory.StartNew(() => MyFunction(progress), TaskCreationOptions.LongRunning); 
    TaskList.Add(task); 
} 

這是陷入用戶界面。有沒有更好的方法來實現這個功能，同時保持UI的響應？

有人建議在使用TPL數據流提出的意見，但由於時間的限制和規範，替代解決方案，歡迎

更新

我不知道進展情況報告是否可能是問題？這裏是什麼樣子：

private IProgress<string> Progress() 
{ 
    return new Progress<string>(msg => 
    { 
     txtMsg.AppendText(msg); 
    }); 
} 

Answer 1

現在，只要任務完成後，我想啓動一個新的。從本質上講，這個過程應該無限期地運行，其在任何給定的時間

這聽起來我跑X的任務，比如你想要一個無限循環內你的任務：

for (int i = 0; i < X; i++) 
{ 
    var progress = Progress(); // returns a new IProgress<T> 
    var task = RunIndefinitelyAsync(progress); 
    TaskList.Add(task); 
} 

private async Task RunIndefinitelyAsync(IProgress<T> progress) 
{ 
    while (true) 
    { 
    try 
    { 
     await Task.Run(() => MyFunction(progress)); 
     // handle success 
    } 
    catch (Exception ex) 
    { 
     // handle exceptions 
    } 
    // update some labels/textboxes in the UI 
    } 
} 

然而，我懷疑「陷入用戶界面」可能在// handle success和/或// handle exceptions代碼中。如果我的懷疑是正確的，那麼儘可能多地將邏輯推入Task.Run。

Answer 2

據我所知，您只需要一個並行執行與定義的並行度。有很多方法來實現你想要的。我建議使用阻塞收集和並行類而不是任務。

所以，當用戶點擊按鈕時，你需要創建一個新的阻塞集合，這將是您的數據源：

BlockingCollection<IProgress> queue = new BlockingCollection<IProgress>(); 
CancellationTokenSource source = new CancellationTokenSource(); 

現在你需要一個亞軍，這將並行執行的：

Task.Factory.StartNew(() => 
    Parallel.For(0, X, i => 
    { 
     foreach (IProgress p in queue.GetConsumingEnumerable(source.Token)) 
     { 
      MyFunction(p); 
     } 
    }), source.Token); 

或者你可以選擇更正確的方式與分區。所以，你需要一個分區類：

private class BlockingPartitioner<T> : Partitioner<T> 
{ 
    private readonly BlockingCollection<T> _Collection; 
    private readonly CancellationToken _Token; 

    public BlockingPartitioner(BlockingCollection<T> collection, CancellationToken token) 
    { 
     _Collection = collection; 
     _Token = token; 
    } 

    public override IList<IEnumerator<T>> GetPartitions(int partitionCount) 
    { 
     throw new NotImplementedException(); 
    } 

    public override IEnumerable<T> GetDynamicPartitions() 
    { 
     return _Collection.GetConsumingEnumerable(_Token); 
    } 

    public override bool SupportsDynamicPartitions 
    { 
     get { return true; } 
    } 
} 

和亞軍將是這樣的：

ParallelOptions Options = new ParallelOptions(); 
Options.MaxDegreeOfParallelism = X; 

Task.Factory.StartNew(
    () => Parallel.ForEach(
     new BlockingPartitioner<IProgress>(queue, source.Token), 
     Options, 
     p => MyFunction(p))); 

所以你現在需要的是填補queue必要的數據。你可以隨時做到。

和最後的觸摸，當用戶取消操作時，你有兩個選擇：

• 首先，你可以打破執行與source.Cancel電話，
• 也可以優雅地通過標記收集完整的停止執行（queue.CompleteAdding）在這種情況下，跑步者將執行所有已排隊的數據並完成。

當然，您需要額外的代碼來處理異常，進度，狀態等。但主要想法在這裏。