簡單的並行任務及延續

Question

我一直在閱讀2個小時，我仍然感到困惑。有人說使用StartNew，有人說Task.Run有人說別的。我知道Task.Run給我一個編譯錯誤。

我需要並行啓動多個任務，然後當每個任務成功完成時執行一個延續任務。知道什麼時候所有的阻塞都會有幫助。

以下是我有：

public void DoSomeWork(object workItem) 
    { 
     var tasks = new Task<ResultArgs>[_itemList.Count]; 

     for (int loopCnt = 0; loopCnt < _itemList.Count; loopCnt++) 
     { 
      tasks[loopCnt] = new Task<ResultArgs>.Run(() => 
      { 
       return _itemList[loopCnt].Analyze(workItem); 
      }); 
      tasks[loopCnt].ContinueWith(ReportResults, TaskContinuationOptions.ExecuteSynchronously); 
     } 
    } 

的編譯表示運行中不存在的任務。

很明顯，我有一些東西跑，但我不知道是什麼。

我該如何解決這個問題？

Answer 1

您可以使用async方法，也可以將項目流入數據流，以下代碼使用Tpl-dataflow來處理項目，將它們傳遞到第二個處理步驟，最後等待處理完成。

using NUnit.Framework; 
using System; 
using System.Collections.Concurrent; 
using System.Linq; 
using System.Threading.Tasks; 
using System.Threading.Tasks.Dataflow; 

namespace AsyncProcessing { 

    [TestFixture] 
    public class PipelineTests { 

     [Test] 
     public async Task RunPipeline() { 
      var pipeline = new MyPipeline(); 
      var data = Enumerable.Range(0, 1000).Select(x => new WorkItem(x, x)); 

      foreach(var item in data) { 
       await pipeline.SendAsync(item); 
      } 

      pipeline.Complete(); 
      await pipeline.Completion; 

      //all processing complete    
     } 
    } 

    class MyPipeline { 

     private BufferBlock<WorkItem> inputBuffer; 
     private TransformBlock<WorkItem, WorkItem> analyzeBlock; 
     private TransformBlock<WorkItem, ResultArg> reportBlock; 
     private ActionBlock<ResultArg> postOutput; 

     public ConcurrentBag<ResultArg> OutputBuffer { get; } 
     public Task Completion { get { return postOutput.Completion; } } 

     public MyPipeline() { 
      OutputBuffer = new ConcurrentBag<ResultArg>(); 
      CreatePipeline(); 
      LinkPipeline(); 
     } 

     public void Complete() { 
      inputBuffer.Complete(); 
     } 

     public async Task SendAsync(WorkItem data) { 
      await inputBuffer.SendAsync(data); 
     } 

     public void CreatePipeline() { 
      var options = new ExecutionDataflowBlockOptions() { 
       MaxDegreeOfParallelism = Environment.ProcessorCount, 
       BoundedCapacity = 10 
      }; 

      inputBuffer = new BufferBlock<WorkItem>(options); 

      analyzeBlock = new TransformBlock<WorkItem, WorkItem>(item => { 
       //Anylyze item.... 
       return item; 
      }, options); 

      reportBlock = new TransformBlock<WorkItem, ResultArg>(item => { 
       //report your results, email.. db... etc. 
       return new ResultArg(item.JobId, item.WorkValue); 
      }, options); 

      postOutput = new ActionBlock<ResultArg>(item => { 
       OutputBuffer.Add(item); 
      }, options); 
     } 

     public void LinkPipeline() { 
      var options = new DataflowLinkOptions() { 
       PropagateCompletion = true, 
      }; 

      inputBuffer.LinkTo(analyzeBlock, options); 
      analyzeBlock.LinkTo(reportBlock, options); 
      reportBlock.LinkTo(postOutput, options); 
     } 
    } 

    public class WorkItem { 

     public int JobId { get; set; } 
     public int WorkValue { get; set; } 

     public WorkItem(int id, int workValue) { 
      this.JobId = id; 
      this.WorkValue = workValue; 
     } 
    } 

    public class ResultArg { 

     public int JobId { get; set; } 
     public int Result { get; set; } 

     public ResultArg(int id, int result) { 
      this.JobId = id; 
      this.Result = result; 
     } 
    } 
} 

Answer 2

爲什麼不使用Parallel.ForEach循環。這是用於並行執行的任務，它可以使用多個線程和執行更快 Parallet.Foreach

但是，如果您正在做一些涉及鎖定的數據庫相關的輸入輸出操作，它可能會失敗。在這種情況下，我會建議在每個任務中保留一個返回類型，並根據前一個任務的返回類型啓用ext任務。