.NET任务返回对象并在内部调用异步

Question

当all（）>时，我有一些在中执行的任务。如果任务返回对象并在其run（）中调用异步方法，我会遇到语义错误。异步方法从blob获取一些字符串内容，然后构造并返回对象。

您知道如何解决此问题，同时维护任务完成的批次下载？ 我需要一个带有finalWrapperObjects的列表。

错误信息

不能将异步lamba表达式转换为委托类型 'func＆lt; finalWrapperObject＆gt;'。异步的lambda表达可能会返回 无效，任务或任务，它们都不可转换为 'func＆lt; finalWrapperObject＆gt;'。

...
List<FinalWrapperObject> finalReturns = new List<FinalWrapperObject>();
List<Task<FinalWrapperObject>> tasks = new List<Task<FinalWrapperObject>>();
var resultsBatch = fetchedObjects.Skip(i).Take(10).ToList();

foreach (var resultBatchItem in resultsBatch)
{
    tasks.Add(
        new Task<FinalWrapperObject>(async () => //!! errors here on arrow
        {
            var blobContent = await azureBlobService.GetAsync(resultBatchItem.StoragePath);
            return new FinalWrapperObject {
                BlobContent = blobContent,
                CreationDateTime = resultBatchItem.CreationDateTime
            };
        })
    );
}

FinalWrapperObject[] listFinalWrapperObjects = await Task.WhenAll(tasks);
finalReturns.AddRange(listFinalWrapperObjects);

return finalReturns;

Answer 1

您的代码永远不会启动任何任务。无论如何，任务不是线程。他们是承诺，即某些东西将完成并可能在未来 Future 中产生价值。有些任务需要线程运行。这些是使用来自ThreadPool的线程执行的。其他，例如异步IO操作，不需要线程。上传文件就是这样的IO操作。

您的lambda是异步的，并且已经返回任务，因此没有理由使用task.run。您可以为所有项目执行一次，收集列表中的任务并等待所有项目。这就是裸露的方式：

async Task<FinalWrapperObject> UploadItemAsync(BatchItem resultBatchItem) =>
{
    var blobContent = await azureBlobService.GetAsync(resultBatchItem.StoragePath);
    return new FinalWrapperObject {
        BlobContent = blobContent,
        CreationDateTime = resultBatchItem.CreationDateTime
    };
}

...

var tasks=resultsBatch.Select(UploadItemAsync);
var results=await Task.WhenAll(tasks);

使用TPL DataFlow

一个更好的选择是使用TPL DataFlow类并同时上传项目，甚至通过处理块中的管道构造管道。

var options= new ExecutionDataflowBlockOptions
         {
            MaxDegreeOfParallelism = 10
         };
var results=new BufferBlock<FinalWrapperObject>();
var uploader=new TransformBlock<BatchItem,FinalWrapperObject>(UploadItemAsync,options);
uploader.LinkTo(results);

foreach(var item in fetchedObjects)
{
    uploader.PostAsync(item);
}
uploader.Complete();
await uploader.Completion;

默认情况下，一个块一次仅处理一条消息。使用MaxDegreeofParallelism = 10我们告诉它同时处理10个项目。只要有项目要发布到uploader块，此代码将同时上传10个项目。

结果将转发到结果 BufferBlock。这些项目可以使用tryReceiveAll：

IList<FinalWrapperObject> items;
results.TryReceiveAll(out items);

数据流块可以合并到管道中。您可以拥有一个从磁盘加载项目的块，另一个可以上传它们，最后一个存储对另一个文件或数据库的响应的块：

var dop10= new ExecutionDataflowBlockOptions
         {
            MaxDegreeOfParallelism = 10,
            BoundedCapacity=4
         };
var bounded= new ExecutionDataflowBlockOptions
         {
            BoundedCapacity=4
         };

var loader=new TransformBlock<FileInfo,BatchItem>(LoadFile,bounded);
var uploader=new TransformBlock<BatchItem,FinalWrapperObject>(UploadItemAsync,dop10);
var dbLogger=new ActionBlock<FinalWrapperObject>(bounded);

var linkOptions=new DataflowLinkOptions {PropagateCompletion=true};
loader.LinkTo(uploader,linkOptions);
uploader.LinkTo(dbLogger,linkOptions);

var folder=new DirectoryInfo(rootPath);
foreach(var item in folder.EnumerateFiles())
{
    await loader.SendAsync(item);
}
loader.Complete();

await dbLogger.Completion;

在这种情况下，文件夹中的所有文件都发布到loader块一个一个一个一个一个a and Fornef的文件batchitem。 uploader上传文件，结果由dblogger存储。最后，我们告诉loader我们已经完成了，并等待所有项目都可以一直处理到等待dblogger.completion。

boundedcapacity用于对每个块的输入缓冲区可以保存多少个项目。这样可以防止将所有文件加载到内存中。