c# - TPL 数据流转换 block 并行发布到两个操作 block

Question

我有一个 TPL 数据流，仅使用转换 block 和操作 block 就可以正常工作。我添加了一个新的操作 block ，与现有的操作 block 同时执行，但我的新操作 block 永远不会被命中。没有抛出任何错误或异常。我需要在代码中添加一个步骤吗？

var ListDocId = new ConcurrentBag<string>(ConvertDataSetToList(IdDocDataSet));

if (ListDocId.Any())
{
    var num_thread = GetThreadNumber();

    //Initialize the pipeline of actions
    var downloadBlock = new TransformBlock<string, RequestObject>(docId =>
        new RequestObject
        {
            DownloadedFile = ListDownloadocId),
            IdDoc = docId
        },
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
    );

    var uploadInS3Block = new ActionBlock<S3RequestUpload>(requestS3Upload =>
        UploadFileAsync(RequestObject.DownloadedFile, RequestObject.IdDoc),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
    );

    var InsertdocIdIntoDbBlock = new ActionBlock<RequestObject>(s3Request =>
        InsertIntotDataBase(s3Request.IdDoc, InsertDate),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
    );

    var options = new DataflowLinkOptions { PropagateCompletion = true };

    downloadBlock.LinkTo(uploadInS3Block, options);
    downloadBlock.LinkTo(InsertdocIdIntoDbBlock, options);

    foreach (var idDoc in ListDocId)
        downloadInAsterionBlock.Post(idDoc);

    downloadBlock.Complete();
    //uploadInS3Block.Completion.Wait();
    //InsertdocIdIntoDbBlock.Completion.Wait();
    Task.WhenAll(uploadInS3Block.Completion,
        InsertdocIdIntoDbBlock.Completion).Wait();

Answer 1

您不能将一个 TransformBlock 链接到多个其他 block 。因此，仅考虑第一次调用 downloadInAsterionBlock.LinkTo()。

您需要在 downloadBlock 和两个 ActionBlock block 之间放置一个 BroadcastBlock。

downloadBlock -> broadcastBlock -> uploadInS3Block
                                -> InsertdocIdIntoDbBlock 

在代码中它看起来像这样:

var bc = new BroadcastBlock<RequestObject>(ro => ro);

downloadBlock.LinkTo(bc);
bc.LinkTo(uploadInS3Block);
bc.LinkTo(InsertdocIdIntoDbBlock);