継続的な単純な並列タスク

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を使用してアイテムを処理し、2番目の処理ステップに渡して最後に処理の完了を待ちます。

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ループを使用しないでください。これは並列実行タスクに使用され、複数のスレッドを使用でき、実行は高速です。その場合、私は各タスクで戻り値の型を保持し、前のタスクの戻り値の型に基づいてextタスクを有効にすることを提案します。