想要使用带有进度报告的任务并行库来更新数据库

Question

我开发了一个存储多个连接字符串的应用程序。我只是迭代for循环并连接每个数据库并对每个数据库执行sql。通过这种方式，我使用批量 sql 语句更新多个数据库。

现在我需要使用任务并行库来同时更新多个数据库，而不是一个接一个地循环更新。我也想处理异常，也想显示多个数据库操作的多个进度条。应该有暂停和恢复功能。

当我单击按钮时，将打开多个数据库连接，并且每个任务都会在我的表单中添加一个新的进度条。每个进度条都会显示每个 db 操作的进度。当任何任务完成时，相应的进度条将从表单中删除。

任何人都可以通过示例代码指导我如何使用 TPL 进行操作。 在这里，我得到了一个更新一个进度条的代码，但我需要更新多个进度条。 int 迭代次数 = 100；

ProgressBar pb = new ProgressBar();   
pb.Maximum = iterations;   
pb.Dock = DockStyle.Fill;   
Controls.Add(pb);   

Task.Create(delegate   
{   
    Parallel.For(0, iterations, i =>  
    {   
        Thread.SpinWait(50000000); // do work here   
        BeginInvoke((Action)delegate { pb.Value++; });   
    });   
}); 

更新问题

我就是这样做的。代码有效，但所有进度条值按顺序增加。 我在 winform 应用程序中有一个表单和一个用户控件。请查看我的代码并告诉我那里有什么问题。

主要代码

public partial class Main : Form
    {
        public Main()
        {
            InitializeComponent();
            this.DoubleBuffered = true;
        }

        private void btnStart_Click(object sender, EventArgs e)
        {
            Progress ucProgress = null;
            Dictionary<string, string> dicList = new Dictionary<string, string>();
            dicList.Add("GB", "conn1");
            dicList.Add("US", "conn2");
            dicList.Add("DE", "conn3");
            fpPanel.Controls.Clear();

            Task.Factory.StartNew(() =>
            {
                foreach (KeyValuePair<string, string> entry in dicList)
                {
                    ucProgress = new Progress();
                    ucProgress.Country = entry.Key;
                    ucProgress.DBConnection = entry.Value;

                    fpPanel.BeginInvoke((MethodInvoker)delegate
                    {
                        fpPanel.Controls.Add(ucProgress);
                        ucProgress.Process();
                    });
                    //fpPanel.Controls.Add(ucProgress);


                    System.Threading.Thread.SpinWait(5000000);
                }
            });

        }

        private void Main_Resize(object sender, EventArgs e)
        {
            this.Invalidate();
        }
        private void Main_Paint(object sender, PaintEventArgs e)
        {
            using (LinearGradientBrush brush = new LinearGradientBrush(this.ClientRectangle,
                                                               Color.WhiteSmoke,
                                                               Color.LightGray,
                                                               90F))
            {
                e.Graphics.FillRectangle(brush, this.ClientRectangle);
            }
        }
    }

用户控制代码

public partial class Progress : UserControl
    {
        public Progress()
        {
            InitializeComponent();
            lblMsg.Text = "";
            pbStatus.Minimum = 0;
            pbStatus.Maximum = 100;
        }

        public string Country { get; set; }
        public string DBConnection { get; set; }
        public string Sql { get; set; }

        public void SetMessage(string strMsg)
        {
            lblMsg.Text = strMsg;
        }

        public void Process()
        {
            var uiScheduler = TaskScheduler.FromCurrentSynchronizationContext();
            Task.Factory.StartNew(() =>
            {
                lblMsg.BeginInvoke((MethodInvoker)delegate
                {
                    lblMsg.Text = "Connecting country " + Country;
                });

                pbStatus.BeginInvoke((MethodInvoker)delegate
                {
                    pbStatus.Value = 30;
                });
                System.Threading.Thread.SpinWait(50000000);

                //***********
                lblMsg.BeginInvoke((MethodInvoker)delegate
                {
                    lblMsg.Text = "executing sql for country " + Country;
                });

                pbStatus.BeginInvoke((MethodInvoker)delegate
                {
                    pbStatus.Value = 60;
                });
                System.Threading.Thread.SpinWait(50000000);

                //***********

                lblMsg.BeginInvoke((MethodInvoker)delegate
                {
                    lblMsg.Text = "sql executed successfully for country " + Country;
                });

                pbStatus.BeginInvoke((MethodInvoker)delegate
                {
                    pbStatus.Value = 100;
                });
                System.Threading.Thread.SpinWait(50000000);

            });
            //System.Threading.Thread.SpinWait(50000000); // do work here   
        }
    }

Answer 1

也许它可以作为起点。处理暂停/恢复取决于您的需要，并且可以进行调整。

var cancellationTokenSource = new CancellationTokenSource();
var cancellation = cancellationTokenSource.Token;

void UpdateDatabases(IEnumerable<...> databases, CancellationToken cancellation)
{
   foreach(db in databases)
   {


   //create as many ProgressBar instances as databases you want to update
   //check if ProgressBar exist, then return it and reuse, otherwise create new
   ProgressBar pb = new ProgressBar();   
   pb.Maximum = iterations;   
   pb.Dock = DockStyle.Fill;   

   Controls.Add(pb);  



  //start thread for every database/progress bar

  Task.Factory.StartNew(progressBar => 
  {   
      var start = (ProgressBar)progressBar).Value; //use last value in case of pause
      Parallel.For(start, iterations, 
          new ParallelOptions(){CancellationToken =  cancellation}
      (i, loopState) =>  
      {   
          if (loopState.ShouldExitCurrentIteration)
                return;
          //perhaps check loopState.ShouldExitCurrentIteration inside worker method
          Thread.SpinWait(50000000); // do work here   

          BeginInvoke((Action)delegate { ((ProgressBar)progressBar).Value++; });   
      });   
   }, 
   pb, cancellation)

   .ContinueWith(task => 
  {
      //to handle exceptions use task.Exception member

      var progressBar = (ProgressBar)task.AsyncState;
      if (!task.IsCancelled)
      {
          //hide progress bar here and reset pb.Value = 0
      }
  }, 
  TaskScheduler.FromCurrentSynchronizationContext() //update UI from UI thread
  ); 

   }
}

 //.........

 //Call
 UpdateDatabases(databases, cancellation)  

 //To suspend, call 
 cancellationTokenSource.Cancel();

 //To resume - simply call UpdateDatabases  again
 cancellationTokenSource = new CancellationTokenSource();
 cancellation = cancellationTokenSource.Token;
 UpdateDatabases(databases, cancellation)  

更新

我已经查看了您的代码。查看重新访问的代码并根据您的需要进行调整。主要错误 - 与closures 混淆并从非 ui 线程创建进度。要启用并行处理，您可以使用 Parallel.ForEach（请参阅 MSND 了解可能的重载）。而且这个设计对我来说看起来有点奇怪（你在 Progress 中混合了数据和逻辑）。从 UI 的角度来看，进度条会按处理顺序显示，而不是按照它们在列表中的原始顺序显示，这也很奇怪（如果您决定按字母顺序对列表进行排序，这将是一个问题）

希望对你有帮助。

主代码

   private void btnStart_Click(object sender, EventArgs e)
    {
        Progress ucProgress = null;
        Dictionary<string, string> dicList = new Dictionary<string, string>();
        dicList.Add("GB", "conn1");
        dicList.Add("US", "conn2");
        dicList.Add("DE", "conn3");
        fpPanel.Controls.Clear();

        Func<KeyValuePair<string, string>, object> createProgress = entry =>
        {

            var tmp = new Progress {Country = entry.Key, DBConnection = entry.Value};
            fpPanel.Controls.Add(tmp);
            return tmp;
        };

        Task.Factory.StartNew(() =>
        {
            //foreach (KeyValuePair<string, string> entry in dicList)

            Parallel.ForEach(dicList,
                entry =>
                {

                    //create and add the Progress in UI thread
                    var ucProgress = (Progress)fpPanel.Invoke(createProgress, entry);

                    //execute ucProgress.Process(); in non-UI thread in parallel. 
                    //the .Process(); must update UI by using *Invoke
                    ucProgress.Process();

                    System.Threading.Thread.SpinWait(5000000);
                });
        });

    }

用户控制代码

public void Process()
    {
        //uiScheduler - Not used
        //var uiScheduler = TaskScheduler.FromCurrentSynchronizationContext();
        //The Task is not necessary because the Process() called from Parallel.ForEach 
        //Task.Factory.StartNew(() =>
        //{

            //BeginInvoke or Invoke required
            lblMsg.BeginInvoke((MethodInvoker)delegate
            {
                lblMsg.Text = "Connecting country " + Country;
            });

            pbStatus.BeginInvoke((MethodInvoker)delegate
            {
                pbStatus.Value = 30;
            });
            System.Threading.Thread.SpinWait(50000000);

            //***********
            lblMsg.BeginInvoke((MethodInvoker)delegate
            {
                lblMsg.Text = "executing sql for country " + Country;
            });

            pbStatus.BeginInvoke((MethodInvoker)delegate
            {
                pbStatus.Value = 60;
            });
            System.Threading.Thread.SpinWait(50000000);

            //***********

            lblMsg.BeginInvoke((MethodInvoker)delegate
            {
                lblMsg.Text = "sql executed successfully for country " + Country;
            });

            pbStatus.BeginInvoke((MethodInvoker)delegate
            {
                pbStatus.Value = 100;
            });
            System.Threading.Thread.SpinWait(50000000);

        //});
        //System.Threading.Thread.SpinWait(50000000); // do work here   
    }