.net - TPL数据流: creating a custom splitblock

Question

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

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

根据我的需要，将输入拆分为两个不同的输出应该足够了。类头应该看起来像这样......

public abstract class SplitBlock<TInput, TOutputLeft, TOutputRight>

我的问题是我真的不知道如何继续。我所知道的是我需要两个 TransformBlocks:

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

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

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

Answer 1

我会首先考虑该类的通用接口(interface)应该是什么样子。我认为最简单的解决方案是这样的:

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

这样，实现就很自然了:一个输入 block 连接到两个输出 block 。唯一的问题是实际处理是否应该在输出 block 中完成(就像您使用两个 TransformBlock 所建议的那样)还是在输入 block 中完成。

如果您想在输出 block 中进行处理，输入 block 可以是一个 ActionBlock，它将输入发送到两个输出，并且输出将是 TransformBlock，正如你所建议的。

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();
            });
    }
}

(这假设左变换和右变换可以同时处理相同的输入。)

另一方面，如果您想在输入 block 中执行处理(这对我来说更有意义)，那么您可以将 ActionBlock 作为输入，并使用 BufferBlock 作为输出，输入 block 处理输入，然后将结果发送到输出 block :

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)))
    {}
}