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()。

您需要在downloadBlock 和两个ActionBlock 块之间添加一个BroadcastBlock。

downloadBlock -> broadcastBlock -> uploadInS3Block
                                -> InsertdocIdIntoDbBlock 

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

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

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