TPL 数据流，JoinBlock 限制的替代方案？

Question

我寻找 JoinBlock 的替代方案，它可以通过 n-TransformBlocks 链接，并将所有 TransformBlock 源块的消息连接/合并在一起，以便将此类的集合传递给另一个数据流块。

JoinBlock 可以很好地完成工作，但仅限于连接 3 个源块。它还存在许多效率低下的问题（连接 2 个源块的偶数值类型（整数）非常慢）。有没有办法让任务从 TransformBlocks 返回并等到所有 TransformBlocks 都有一个完成的任务才能接受Task<item>？

任何替代想法？我可能有 1-20 个这样的转换块，在传递连接的项目集合之前，我需要将哪些项目连接在一起。每个转换块都保证为每个“转换”的输入项准确返回一个输出项。

编辑：要求澄清：

根据我之前的一个问题，我按如下方式设置了我的 JoinBlock：

public Test()
{
    broadCastBlock = new BroadcastBlock<int>(i =>
        {
            return i;
        });

    transformBlock1 = new TransformBlock<int, int>(i =>
        {
            return i;
        });

    transformBlock2 = new TransformBlock<int, int>(i =>
        {
            return i;
        });

    joinBlock = new JoinBlock<int, int>();

    processorBlock = new ActionBlock<Tuple<int, int>>(tuple =>
        {
            //Console.WriteLine("tfb1: " + tuple.Item1 + "tfb2: " + tuple.Item2);
        });

    //Linking
    broadCastBlock.LinkTo(transformBlock1, new DataflowLinkOptions { PropagateCompletion = true });
    broadCastBlock.LinkTo(transformBlock2, new DataflowLinkOptions { PropagateCompletion = true });
    transformBlock1.LinkTo(joinBlock.Target1);
    transformBlock2.LinkTo(joinBlock.Target2);
    joinBlock.LinkTo(processorBlock, new DataflowLinkOptions { PropagateCompletion = true });
}

public void Start()
{
    Stopwatch watch = new Stopwatch();
    watch.Start();

    const int numElements = 1000000;

    for (int i = 1; i <= numElements; i++)
    {
        broadCastBlock.Post(i);
    }

    ////mark completion
    broadCastBlock.Complete();
    Task.WhenAll(transformBlock1.Completion, transformBlock2.Completion).ContinueWith(_ => joinBlock.Complete());


    processorBlock.Completion.Wait();

    watch.Stop();

    Console.WriteLine("Time it took: " + watch.ElapsedMilliseconds + " - items processed per second: " + numElements / watch.ElapsedMilliseconds * 1000);
    Console.ReadLine();
}

Answer 1

一种方法是使用BatchBlockwith Greedyset to false。在此配置中，块不会执行任何操作，直到有n来自n不同块的项目等待它被消耗（其中n是您在创建 时设置的数量BatchBlock）。发生这种情况时，它会n立即消耗所有项目并生成一个包含所有项目的数组。

此解决方案的一个警告是结果数组未排序：您不会知道哪个项目来自哪个来源。而且我不知道它的性能与 相比如何JoinBlock，您必须自己进行测试。（虽然我会理解使用BatchBlock这种方式是否会更慢，因为非贪婪消费所需的开销。）