TPL 数据流转换块并行发布到两个操作块

Question

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

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到多个其他块。所以只考虑第一次调用downloadInAsterionBlock.LinkTo()。

您需要在和 两个块BroadcastBlock之间放置一个。downloadBlockActionBlock

downloadBlock -> broadcastBlock -> uploadInS3Block
                                -> InsertdocIdIntoDbBlock 

在代码中它将如下所示：

downloadBlock -> broadcastBlock -> uploadInS3Block
                                -> InsertdocIdIntoDbBlock