我已经读了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);
}
}
编译说任务中不存在Run。
显然,我正在运行一些东西,但我不知道是什么。
我该如何克服这个问题?
您可以使用
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;
}
}
}
为什么不使用 Parallel.ForEach 循环。这用于并行执行任务,它可以使用多个线程并且执行速度更快 Parallet.Foreach
但是,如果您正在执行一些涉及锁定的与数据库相关的输入输出操作,则可能会失败。在这种情况下,我建议在每个任务中保留一个返回类型,并根据前一个任务的返回类型启用扩展任务。