TPL 数据流 C# 等待所有链接块完成

Question

我正在使用 TPL 数据流构建管道。此管道在逻辑上应执行以下操作：

1. 首先处理多个数据项 - 假设是 pollingBlock。
2. 如果满足某些条件，将一个项（满足条件的）传递到特定块以进行进一步监控，假设它是monitoringBlock。每个monitoringBlock只能放1个物品，但是有多个monitoringBlocks.
3. pollingBlock 应继续处理所有项目，包括以 while (true) 方式发布的项目。
4. monitoringBlocks 在被占用时不应接受任何其他消息，这些消息应该直接删除而无需进一步处理。
5. 在 monitoringBlock 中进行一些处理后，消息应标记为已完成或转移到下一个块进行处理，下一个块是 processingBlock

一个简短的示例：

public Task ExecutePipeline()
{
    var block = CreatePollingPipeline();
    block.Post((_serviceOne, _serviceTwo));

    block.Complete();
    return block.Completion;
}

public ActionBlock<(IServiceOne serviceOne, IServiceTwo serviceTwo)> CreatePollingPipeline()
{
    var pollingAlertHolder = new BufferBlock<(string input1, string input2)>();

    var pollingBlock = new ActionBlock<(IServiceOne serviceOne, IServiceTwo serviceTwo)>(services =>
    {
        while (true)
        {
            Console.WriteLine("Posting to alert block");
            pollingAlertHolder.Post(("INP1", "INPVAL"));
            Thread.Sleep(2000);

            Console.WriteLine("Posting to alert block");
            pollingAlertHolder.Post(("INP1", "INPVAL"));
            Thread.Sleep(2000);

            Console.WriteLine("Posting to alert block");
            pollingAlertHolder.Post(("INP2", "INPVAL2"));
            Thread.Sleep(2000);

            Console.WriteLine("Posting to alert block");
            pollingAlertHolder.Post(("INP1", "INPVAL"));
            Thread.Sleep(2000);

            Console.WriteLine("Posting to alert block");
            pollingAlertHolder.Post(("INP1", "INPVAL"));
            Thread.Sleep(2000);

            Console.WriteLine("Posting to alert block");
            pollingAlertHolder.Post(("INP2", "INPVAL2"));
            Thread.Sleep(2000);
        }
    });

    var monitoringBlock = new TransformBlock<(string input1, string input2), (string input1, string input2)>(inputs =>
        {
            Console.WriteLine("monitoringBlock started");
            Thread.Sleep(5000);
            Console.WriteLine("monitoringBlock completed");

            return (inputs.input1, inputs.input2);
        },
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, BoundedCapacity = 1 });

    pollingAlertHolder.LinkTo(monitoringBlock, new DataflowLinkOptions() { PropagateCompletion = true },
        inputs => inputs.input1 == "INP1" && inputs.input2 == "INPVAL");
    pollingAlertHolder.LinkTo(DataflowBlock.NullTarget<(string input1, string input2)>());

    var processingBlock = new ActionBlock<(string input1, string input2)>(i =>
    {
        Console.WriteLine("processingBlock started");
        Thread.Sleep(2000);
        Console.WriteLine("processingBlock completed");
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, BoundedCapacity = 1 });
    monitoringBlock.LinkTo(processingBlock, new DataflowLinkOptions { PropagateCompletion = true });


    return pollingBlock;
}

我的问题是如何让 monitoringBlock 占用直到链接的 processingBlock 完成它的工作？我不希望在消息完成 FULL 处理周期之前将任何项目发布到 monitoringBlock。

Answer 1

评论中已经提到，你可以简单地将monitoringBlock和processingBlock的逻辑封装在一个块中，例如你可以通过预定义的Datablock.Encapsulate方法来实现。

然而，如果你不想那样做，你可以使用 AutoResetEvent 或类似的抽象，你的代码可以是这样的：

AutoResetEvent dataflowEvent = new AutoResetEvent(true);
var bufferBlock = new ActionBLock<(string input1, string input2)>(i =>
{
    dataflowEvent.WaitOne();
    monitoringBlock.Post(i);
});
var monitoringBlock = new TransformBlock<(string input1, string input2), (string input1, string input2)>(inputs =>
    {
        Console.WriteLine("monitoringBlock started");
        Thread.Sleep(5000);
        Console.WriteLine("monitoringBlock completed");

        dataflowEvent.Set();
        return (inputs.input1, inputs.input2);
    },
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, BoundedCapacity = 1 });