TPL DataFlow:创建自定义拆分块

Question

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

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

根据我的需要,它应该足以将输入分成两个不同的输出.类头应该看起来像这样......

public abstract class SplitBlock<TInput, TOutputLeft, TOutputRight>

我的问题是我不知道该怎么做.我所知道的是我需要两个TransformBlocks:

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

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

我很感激你能给予的每一个暗示.

Answer 1

I would start with thinking about how the general interface of that class should look like. I think the simplest solution would be something like:

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

With that, the implementation follows naturally: one input block connected to two output blocks. The only question is whether the actual processing should be done in the output blocks (like you suggested with your two TransformBlocks) or in the input block.

If you want to have processing in the output blocks, the input block can be an ActionBlock that sends the input to both outputs and the outputs will be TransformBlocks, as you suggested.

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

(This assumes that left and right transforms can process the same input at the same time.)

On the other hand, if you wanted to perform the processing in the input block (which makes more sense to me), then you could have ActionBlock as input and BufferBlocks as outputs, with the input block processing the input and then sending the results to the output blocks:

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