Обновление контроля пользовательского интерфейса из ActionBlock

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 Access и также обновляет пользовательский интерфейс. Но проблема заключается в том, что он блокирует пользовательский интерфейс, который является понятным, поскольку я использую TaskScheduler.FromCurrentSynchronizationContext, который будет обновлять TextBox на тему пользовательского интерфейса.

Я сделал небольшое изменение в коде и удалю планировщик с ExecutionDataflowBlockOptions. Вместо этого я обновить пользовательский интерфейс с помощью следующего кода,

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 

Теперь, что может быть лучшим способом для обновления пользовательского интерфейса в моем сценарии, а также поддержание порядка.

Answer 1

Блоки потока данных TPL сохраняют порядок ввода при их выходе. Они не сохраняют порядок выполнения внутри блока, поэтому вы видите все не в порядке.

Что вы, вероятно, захотите сделать, это заменить ActionBlock на TransformBlock, чтобы фактически выполнить сложную работу и связать ее с ActionBlock, который обновляет пользовательский интерфейс по одному.

Вы также можете иметь этот блок работать на потоке пользовательского интерфейса, так что вы не должны использовать 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 });