TPL DataFlow：创建自定义拆分块

Question

只需要一些帮助，在数据流库的帮助下创建自定义拆分块，该库是 .Net 中 TPL 的一部分。

我想要实现的只是一个简单的自定义块，它接受输入并将其拆分为多个变换块。这是过滤数据所必需的，我可以在其中记录负面条目并继续使用好的条目。

根据我的需要，将输入分成两个不同的输出就足够了。类头应该是这样的......

public abstract class SplitBlock<TInput, TOutputLeft, TOutputRight>

我的问题是我真的不知道该怎么做。我只知道我需要两个 TransformBlock：

var leftBlock  = new TransformBlock<TInput, TOutputLeft>(...)
var rightblock = new TransformBlock<TInput, TOutputRight>(...)

在我所有的尝试中，我最终有多个 ITargetBlocks 用于存储左右块的输入，但这不可能是正确的，可以吗？

感谢您提供的每一个提示。

Answer 1

我会首先考虑该类的通用界面应该是什么样子。我认为最简单的解决方案是：

public class SplitBlock<TInput, TOutputLeft, TOutputRight>
{
    public ITargetBlock<TInput> Input { get; }
    public ISourceBlock<TOutputLeft> LeftOutput { get; }
    public ISourceBlock<TOutputRight> RightOutput { get; }
}

这样，实现自然如下：一个输入块连接到两个输出块。唯一的问题是实际处理应该在输出块中完成（就像你建议的两个TransformBlocks）还是在输入块中。

如果您想在输出块中进行处理，输入块可以是ActionBlock，它将输入发送到两个输出，输出将为TransformBlocks，正如您所建议的那样。

public class SplitBlock<TInput, TOutputLeft, TOutputRight>
{
    private ActionBlock<TInput> input;
    private TransformBlock<TInput, TOutputLeft> leftOutput;
    private TransformBlock<TInput, TOutputRight> rightOutput;

    public ITargetBlock<TInput> Input { get { return input; } }
    public ISourceBlock<TOutputLeft> LeftOutput { get { return leftOutput; } }
    public ISourceBlock<TOutputRight> RightOutput { get { return rightOutput; } }

    public SplitBlock(
        Func<TInput, TOutputLeft> leftTransform,
        Func<TInput, TOutputRight> rightTransform)
    {
        input = new ActionBlock<TInput>(
            x =>
            {
                leftOutput.Post(x);
                rightOutput.Post(x);
            });
        leftOutput = new TransformBlock<TInput, TOutputLeft>(leftTransform);
        rightOutput = new TransformBlock<TInput, TOutputRight>(rightTransform);

        // TODO handle fault in input correctly
        input.Completion.ContinueWith(
            _ =>
            {
                leftOutput.Complete();
                rightOutput.Complete();
            });
    }
}

（假设左右变换可以同时处理相同的输入。）

另一方面，如果您想在输入块中执行处理（这对我来说更有意义），那么您可以将ActionBlock 作为输入，BufferBlocks 作为输出，使用输入块处理输入然后将结果发送到输出块：

public class SplitBlock<TInput, TOutputLeft, TOutputRight>
{
    private ActionBlock<TInput> input;
    private BufferBlock<TOutputLeft> leftOutput;
    private BufferBlock<TOutputRight> rightOutput;

    public ITargetBlock<TInput> Input { get { return input; } }
    public ISourceBlock<TOutputLeft> LeftOutput { get { return leftOutput; } }
    public ISourceBlock<TOutputRight> RightOutput { get { return rightOutput; } }

    public SplitBlock(
        Func<TInput, Tuple<TOutputLeft, TOutputRight>> combinedTransform)
    {
        input = new ActionBlock<TInput>(
            value =>
            {
                var result = combinedTransform(value);
                leftOutput.Post(result.Item1);
                rightOutput.Post(result.Item2);
            });
        leftOutput = new BufferBlock<TOutputLeft>();
        rightOutput = new BufferBlock<TOutputRight>();

        // TODO handle fault in input correctly
        input.Completion.ContinueWith(
            _ =>
            {
                leftOutput.Complete();
                rightOutput.Complete();
            });
    }

    public SplitBlock(
        Func<TInput, TOutputLeft> leftTransform,
        Func<TInput, TOutputRight> rightTransform)
        : this(x => Tuple.Create(leftTransform(x), rightTransform(x)))
    {}
}