从ActionBlock更新UI控件

Question

我一直试图通过创建一个示例应用程序来理解TPL数据流。我一直试图做的一件事是从TextBox更新ActionBlock控件。使用TPL数据流的原因是在保持顺序的同时执行并行异步操作。下面的函数是我写的，

private TaskScheduler scheduler = null;

public Form1()
    {
        this.scheduler = TaskScheduler.FromCurrentSynchronizationContext();
        InitializeComponent();
    }

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

        txtOutput.Clear();

        ExecutionDataflowBlockOptions execOptions = new ExecutionDataflowBlockOptions();
        execOptions.MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded;
        execOptions.TaskScheduler = scheduler;

        ActionBlock<int> actionBlock = new ActionBlock<int>(async v =>
        {
            bool x = await InsertIntoDatabaseAsync(v);

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

        }, execOptions);


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

        actionBlock.Complete();
        await actionBlock.Completion;            

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


private async Task<bool> InsertIntoDatabaseAsync(int id)
    {
        try
        {
            string connString = "Provider=Microsoft.ACE.OLEDB.12.0;Data Source=D:\\TPLDatabase.accdb;Persist Security Info=False;";

            using (OleDbConnection conn = new OleDbConnection(connString))
            {
                string commandText = "INSERT INTO tblRecords (ProductName, ProductDescription, IsProcessed) VALUES (@ProductName, @ProductDescription, @IsProcessed)";

                await conn.OpenAsync();
                using (OleDbCommand command = new OleDbCommand(commandText, conn))
                {
                    command.CommandType = CommandType.Text;

                    command.Parameters.AddWithValue("@ProductName", "Product " + id);
                    command.Parameters.AddWithValue("@ProductDescription", "Description " + id);
                    command.Parameters.AddWithValue("@IsProcessed", false);

                    if (await command.ExecuteNonQueryAsync() > 0)
                        return true;
                    else
                        return false;
                }
            }
        }
        catch
        {
            return false;
        }
    }

现在，上面的代码运行得很好。它在我的示例MS数据库中按顺序插入记录，并按顺序更新UI。但是问题在于它阻塞了UI，这是可以理解的，因为我使用的是TaskScheduler.FromCurrentSynchronizationContext，它将在UI线程上更新TextBox。

我对代码做了一个小小的修改，并将调度程序从ExecutionDataflowBlockOptions中删除。相反，我用以下代码更新UI，

txtOutput.Invoke(new MethodInvoker(delegate
            {
                if (x)
                    txtOutput.Text += "Value Inserted for: " + v + Environment.NewLine;
                else
                   txtOutput.Text += "Value Failed for: " + v + Environment.NewLine;
            }));

现在，这个更改并没有冻结UI，但是数据库中值的顺序和TextBox中显示的值的顺序受到了严重的干扰。新秩序是这样的，

ID   ProductName    ProductDescription  IsProcessed
6847    Product 6     Description 6       False
6848    Product 7     Description 7       False
6849    Product 8     Description 8       False
6850    Product 10    Description 10      False
6851    Product 11    Description 11      False
6852    Product 12    Description 12      False
6853    Product 9     Description 9       False
6854    Product 13    Description 13      False
6855    Product 14    Description 14      False

现在，在我的场景中更新UI并保持顺序的最佳方法是什么。

Answer 1

TPL数据流块在输出时保留输入的顺序。在块内部，它们不保留执行的顺序，这就是为什么您看到所有事情都不正常的原因。

您可能想要做的是用一个ActionBlock替换一个TransformBlock，以实际并发地完成繁重的工作，并将它与一个每次更新UI的ActionBlock链接起来。

您也可以在UI线程上运行这个块，这样就不需要使用Invoke了。

var transformBlock = new TransformBlock<int, int>(
    v => InsertIntoDatabaseAsync(v), 
    execOptions);

var actionBlock = new ActionBlock<int>(x =>
{
    if (x)
        txtOutput.Text += "Value Inserted for: " + v + Environment.NewLine;
    else
        txtOutput.Text += "Value Failed for: " + v + Environment.NewLine;
}, new ExecutionDataflowBlockOptions { TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext() })

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