c# - Parallel.For SendAsync 到 BufferBlock 到异步转换？

Question

我正在学习 TPL Dataflow 库。到目前为止，这正是我所寻找的。

我创建了一个简单的类(如下)，它执行以下功能

• 执行 ImportPropertiesForBranch 后，我会转到第 3 方 API 并获取属性列表
• 返回 xml 列表并将其反序列化为属性数据数组(id、api 端点、lastupdated)。大约有 400 多处特性(如房屋)。
• 然后，我使用 Parallel.For 将属性数据 SendAsync 发送到我的 propertyBufferBlock
• propertyBufferBlock 链接到 propertyXmlBlock(它本身是一个 TransformBlock)。
• 然后，propertyXmlBlock(异步)返回到 API(使用属性数据中提供的 API 端点)并获取属性 xml 进行反序列化。
• 一旦 xml 返回并可用，我们就可以反序列化
• 稍后，我将添加更多 TransformBlock 将其保存到数据存储中。

所以我的问题是；

• 代码中是否存在任何潜在的瓶颈或可能造成麻烦的区域？我知道我没有包含任何错误处理或取消(即将推出)。
• 是否可以在 TransformBlock 内 await 异步调用，或者这是一个瓶颈？
• 虽然代码有效，但我担心 Parallel.For、BufferBlock 和 TransformBlock 中的异步的缓冲和异步性。我不确定这是最好的方法，我可能混淆了一些概念。

欢迎任何指导、改进和陷阱建议。

using System.Diagnostics;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using My.Interfaces;
using My.XmlService.Models;

namespace My.ImportService
{
    public class ImportService
    {

        private readonly IApiService _apiService;
        private readonly IXmlService _xmlService;
        private readonly IRepositoryService _repositoryService;

        public ImportService(IApiService apiService,
            IXmlService xmlService,
            IRepositoryService repositoryService)
        {
            _apiService = apiService;
            _xmlService = xmlService;
            _repositoryService = repositoryService;

            ConstructPipeline();
        }

        private BufferBlock<propertiesProperty> propertyBufferBlock;
        private TransformBlock<propertiesProperty, string> propertyXmlBlock;
        private TransformBlock<string, propertyType> propertyDeserializeBlock;
        private ActionBlock<propertyType> propertyCompleteBlock;

        public async Task<bool> ImportPropertiesForBranch(string branchName, int branchUrlId)
        {
            var propertyListXml = await _apiService.GetPropertyListAsync(branchUrlId);

            if (string.IsNullOrEmpty(propertyListXml))
                return false;

            var properties = _xmlService.DeserializePropertyList(propertyListXml);

            if (properties?.property == null || properties.property.Length == 0)
                return false;

            // limited to the first 20 for testing
            Parallel.For(0, 20,
                new ParallelOptions {MaxDegreeOfParallelism = 3},
                i => propertyBufferBlock.SendAsync(properties.property[i]));

            propertyBufferBlock.Complete();

            await propertyCompleteBlock.Completion;

            return true;
        }

        private void ConstructPipeline()
        {
            propertyBufferBlock = GetPropertyBuffer();
            propertyXmlBlock = GetPropertyXmlBlock();
            propertyDeserializeBlock = GetPropertyDeserializeBlock();
            propertyCompleteBlock = GetPropertyCompleteBlock();

            propertyBufferBlock.LinkTo(
                propertyXmlBlock,
                new DataflowLinkOptions {PropagateCompletion = true});

            propertyXmlBlock.LinkTo(
                propertyDeserializeBlock,
                new DataflowLinkOptions {PropagateCompletion = true});

            propertyDeserializeBlock.LinkTo(
                propertyCompleteBlock,
                new DataflowLinkOptions {PropagateCompletion = true});
        }

        private BufferBlock<propertiesProperty> GetPropertyBuffer()
        {
            return new BufferBlock<propertiesProperty>();
        }

        private TransformBlock<propertiesProperty, string> GetPropertyXmlBlock()
        {
            return new TransformBlock<propertiesProperty, string>(async propertiesProperty =>
                {
                    Debug.WriteLine($"getting xml {propertiesProperty.prop_id}");
                    var propertyXml = await _apiService.GetXmlAsStringAsync(propertiesProperty.url);
                    return propertyXml;
                },
                new ExecutionDataflowBlockOptions
                {
                    MaxDegreeOfParallelism = 1,
                    BoundedCapacity = 2
                });
        }

        private TransformBlock<string, propertyType> GetPropertyDeserializeBlock()
        {
            return new TransformBlock<string, propertyType>(xmlAsString =>
                {
                    Debug.WriteLine($"deserializing");
                    var propertyType = _xmlService.DeserializeProperty(xmlAsString);
                    return propertyType;
                },
                new ExecutionDataflowBlockOptions
                {
                    MaxDegreeOfParallelism = 1,
                    BoundedCapacity = 2
                });
        }

        private ActionBlock<propertyType> GetPropertyCompleteBlock()
        {
            return new ActionBlock<propertyType>(propertyType =>
                {
                    Debug.WriteLine($"complete {propertyType.id}");
                    Debug.WriteLine(propertyType.address.display);
                },
                new ExecutionDataflowBlockOptions
                {
                    MaxDegreeOfParallelism = 1,
                    BoundedCapacity = 2
                });
        }
    }
}

Answer 1

你实际上以错误的方式做了一些事情:

i => propertyBufferBlock.SendAsync(properties.property[i])

您需要 await 方法，否则您将创建太多并发任务。

还有这一行:

MaxDegreeOfParallelism = 1

将限制 block 的执行以进行后续执行，这可能会降低您的性能。

正如您在评论中所说，您切换到同步方法Post，并通过设置BoundedCapacity来限制 block 的容量。应谨慎使用此变体，因为您需要检查它的返回值，该值表明消息是否已被接受。

至于您担心等待 block 内的 async 方法 - 这绝对没问题，并且应该像在 async 方法使用的其他情况下一样完成。