TPL 数据流是并行的或有序的，但不是两者

Question

我最近已经发布了关于 Async、Await、TPL 和 TPL 数据流的各种问题。所有这些问题都得到了解答，让我明白了很多。

我开始研究异步编程，因为我想运行我的异步方法，但有一个问题。我想在任务异步并行运行时保持顺序。记录插入数据库时​​以及使用TextBox 控件在表单上打印时需要维护的顺序。 （这里我必须使用FromCurrentSynchronizationContext，因为我是从UI线程访问控件）

有人建议我使用 TPL 数据流，因为有人告诉我它提供了我需要的功能。经过一番努力并了解了 TPL 数据流的工作原理后，我设法创建了一个简单的应用程序，用于比较同步调用和 TPL 数据流代码之间的性能。

  private void btnStartSync_Click(object sender, EventArgs e)
    {
        Stopwatch watch = new Stopwatch();
        watch.Start();

        try
        {
            txtOutput.Clear();

            for (int i = 1; i <= 200; i++)
            {
                bool x = InsertIntoDatabaseSync(i);

                if (x)
                    txtOutput.Text += "Value Inserted for Id: " + i + Environment.NewLine;
                else
                    txtOutput.Text += "Value Failed for Id: " + i + Environment.NewLine;

                txtOutput.Refresh();
            }

            watch.Stop();
            lblSyncTime.Text = Convert.ToString(watch.ElapsedMilliseconds / 1000);
        }
        catch
        {

        }
    }

上述代码是对同步方法的同步调用，该方法将记录插入数据库。

 public async void btnTPLDataFlow_Click(object sender, EventArgs e)
    {
        Stopwatch watch = new Stopwatch();
        watch.Start();

        txtOutput.Clear();

        ExecutionDataflowBlockOptions execOptions = new ExecutionDataflowBlockOptions();
        execOptions.MaxDegreeOfParallelism = 1;

        var transformBlock = new TransformBlock<string, OutputPropertyClass>(async v =>
        {
            try
            {
                bool x = await InsertIntoDatabaseAsync(Convert.ToInt32(v));

                OutputPropertyClass objResult = new OutputPropertyClass();
                objResult.Id = Convert.ToInt32(v);
                objResult.result = x;

                return objResult;
            }
            catch
            {
                throw new Exception("Exception occurred for: " + v);
            }

        }, execOptions);


        ActionBlock<OutputPropertyClass> actionBlock = new ActionBlock<OutputPropertyClass>(v =>
        {
            try
            {
                if (v.result)
                    txtOutput.Text += "Value Inserted for Id: " + v.Id  + Environment.NewLine;
                else
                    txtOutput.Text += "Value Failed for Id: " + v.Id + Environment.NewLine;
            }
            catch
            {
                throw new Exception("Exception occurred");
            }                

        }, new ExecutionDataflowBlockOptions { TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext() });


        for (int i = 1; i <= 200; i++)
        {
            transformBlock.Post(i.ToString());
        }

        transformBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });
        transformBlock.Complete();

        try
        {
            await transformBlock.Completion;
        }
        catch (AggregateException ex)
        {
            MessageBox.Show(ex.InnerException.Message);
        }
        catch (Exception ex)
        {
            MessageBox.Show(ex.Message);
        }


        watch.Stop();
        lblTPLDataFlow.Text = Convert.ToString(watch.ElapsedMilliseconds / 1000);
    }

上面提到的代码是一个Async调用DataFlow块。 TransformBlock 正在调用 InsertIntoDatabaseAsync 方法，这是一个异步任务。除此之外InsertIntoDatabaseAsync 和InsertIntoDatabaseSync 是相同的。

现在两个调用的完成时间完全相同。在我的系统上，它们都需要 9 秒才能完成。

为了保持TransformBlock 中数据的顺序，我必须将MaxDegreeOfParallelism 限制为1。这样可以保持顺序，但效率没有提高。

如果我尝试增加 MaxDegreeOfParallelism 让我们说 5，那么完成该过程所需的时间会增加，并且超过 25 秒。秩序也受到干扰。

我正在寻找一种同时具有顺序和并行性的方法。如果 TPL Dataflow 无法实现这一点，那么必须有其他方法。请帮助找到正确的方向。

Answer 1

你想要的没有意义。对于流程的每个部分，如果需要订购单个部分，则永远不能并行运行。当您限制所有内容按顺序运行时，它必须一次运行一个。这不是 TPL 数据流的问题，而是需求的问题。

TPL Dataflow 可以将流分成多个部分，每个部分并行运行，并允许每个部分与自身并行运行（使用MaxDegreeOfParallelism > 1），同时保持流本身的顺序。

TPL 数据流保持块的输入和输出顺序，它不保持块内的顺序（将它们插入数据库的位置）。

因此，如果您希望您的所有流程既是并行的又是有序的，TPL 数据流无法帮助您，但没有其他方法可以帮助您。

Answer 2

同时（并行）做很多事情的诀窍是它们不会以相同的顺序完成。要订购一套东西，这套东西需要是完整的，而不是分解成单独的任务。您可能需要在并行任务全部完成后订购该套装。