8

我正在使用 TPL 数据流处理图像。我收到一个处理请求,从流中读取图像,应用几个转换,然后将生成的图像写入另一个流:

Request -> Stream -> Image -> Image ... -> Stream

为此,我使用了这些块:

BufferBlock<Request>
TransformBlock<Request,Stream>
TransformBlock<Stream,Image>
TransformBlock<Image,Image>
TransformBlock<Image,Image>
...
writerBlock = new ActionBlock<Image>

问题是最初Request的内容包含创建结果所需的一些数据Stream以及我当时需要的一些附加信息。我是否必须将原始Request(或其他一些上下文对象)向下传递writerBlock到所有其他块,如下所示:

TransformBlock<Request,Tuple<Request,Stream>>
TransformBlock<Tuple<Request,Stream>,Tuple<Request,Image>>
TransformBlock<Tuple<Request,Image>,Tuple<Request,Image>>
...

(这很丑陋),或者有没有办法将第一个块链接到最后一个块(或者概括地说,链接到需要额外数据的块)?

4

2 回答 2

8

是的,您几乎需要按照您的描述进行操作,将附加数据从每个块传递到下一个块。

但是使用几个辅助方法,您可以使这变得更简单:

public static IPropagatorBlock<TInput, Tuple<TOutput, TInput>>
    CreateExtendedSource<TInput, TOutput>(Func<TInput, TOutput> transform)
{
    return new TransformBlock<TInput, Tuple<TOutput, TInput>>(
        input => Tuple.Create(transform(input), input));
}

public static IPropagatorBlock<Tuple<TInput, TExtension>, Tuple<TOutput, TExtension>>
    CreateExtendedTransform<TInput, TOutput, TExtension>(Func<TInput, TOutput> transform)
{
    return new TransformBlock<Tuple<TInput, TExtension>, Tuple<TOutput, TExtension>>(
        tuple => Tuple.Create(transform(tuple.Item1), tuple.Item2));
}

签名看起来令人生畏,但实际上并没有那么糟糕。

此外,您可能希望添加将选项传递给已创建块的重载,或采用异步委托的重载。

例如,如果您想使用单独的块对数字执行一些操作,同时传递原始数字,您可以执行以下操作:

var source = new BufferBlock<int>();
var divided = CreateExtendedSource<int, double>(i => i / 2.0);
var formatted = CreateExtendedTransform<double, string, int>(d => d.ToString("0.0"));
var writer = new ActionBlock<Tuple<string, int>>(tuple => Console.WriteLine(tuple));

source.LinkTo(divided);
divided.LinkTo(formatted);
formatted.LinkTo(writer);

for (int i = 0; i < 10; i++)
    source.Post(i);

如您所见,您的 lambdas(除了最后一个)仅处理“当前”值(或int,取决于管道的阶段),“原始”值(always )是自动传递的。在任何时候,您都可以使用使用普通构造函数创建的块来访问这两个值(如示例中的 final )。doublestringintActionBlock

(这BufferBlock实际上不是必需的,但我添加它以更接近您的设计。)

于 2013-01-30T20:20:53.353 回答
1

由于我才刚刚开始使用 TPL Dataflow,因此我可能会不知所措。但我相信你可以使用 aBroadcastBlock作为你的源和你的第一个目标之间的中介来实现这一点。

BroadcastBlock可以将消息提供给许多目标,因此您可以使用它来提供给您的目标,以及JoinBlock最后将结果与原始消息合并的a 。

source -> Broadcast ->-----------------------------------------> JoinBlock <source, result>
                    -> Transformation1 -> Transformation 'n'  ->

例如:

var source = new BufferBlock<int>();
var transformation =  new TransformBlock<int, int>(i => i * 100);

var broadCast = new BroadcastBlock<int>(null);
source.LinkTo(broadCast);
broadCast.LinkTo(transformation);

var jb = new JoinBlock<int, int>();
broadCast.LinkTo(jb.Target1);
transformation.LinkTo(jb.Target2);

jb.LinkTo(new ActionBlock<Tuple<int, int>>(
          c => Console.WriteLine("Source:{0}, Target Result: {1}", c.Item1, c.Item2)));

source.Post(1);
source.Post(2);

source.Complete();

产量...

来源:1,目标结果:100

来源:2,目标结果:200

我只是不太确定它在异步环境中的行为方式。

于 2015-07-17T18:21:12.270 回答