c# - TPL数据流，可以替代JoinBlock限制吗？

我正在寻找JoinBlock的替代方法，该方法可以通过n-TransformBlocks链接到，并将所有TransformBlock源块的消息合并/合并在一起，以便将这样的集合传递给另一个数据流块。

JoinBlock可以很好地完成工作，但仅限于最多连接3个源代码块。它也遭受许多低效率的困扰(加入2个源块的偶数类型(ints)非常慢)。有没有办法让Task从TransformBlocks返回并等待所有TransformBlocks都完成一个要传递的任务，然后再接受Task<item>？

还有其他想法吗？我可能有1-20个这样的转换块，在传递联接的项目集合之前，需要将哪些项目联接在一起。保证每个转换块为每个“已转换”输入项恰好返回一个输出项。

编辑:要求澄清:

根据我之前的一个问题，我按如下所示设置了JoinBlocks:

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

最佳答案

一种方法是使用BatchBlock并将Greedy设置为false。在此配置中，该块不会执行任何操作，直到n不同块中有n项等待使用(其中n是您在创建BatchBlock时设置的编号)。发生这种情况时，它将立即使用所有n项，并生成一个包含所有项的数组。

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