并行处理混淆

Question

我对C#编程很陌生。我试图从使用后台工作人员的服务器列表中获取更新的数量。每个服务器的结果显示在报表进度方法的列表视图中。我能够使用foreach循环成功地获得结果，但是当尝试使用并行foreach获得相同的结果时，列表视图的所有列和行都混合在一起。

例如: foreach循环的输出:可用的服务器名称状态更新

1. server1登录服务器失败！0
2. 可获得server2更新3
3. 可获得server3更新3
4. server4，直到现在的0，等等。

平行外汇输出：

1. 可获得server1更新1
2. server1登录服务器失败！1
3. server2登录服务器失败！0
4. server3登录服务器失败！0
5. server4登录服务器失败！0
6. server4更新有3种，等等。

我尝试过锁定部分代码，也尝试过使用并发包，但无法完全解决问题。下面是并行程序代码。我做错什么了吗？任何建议都会有很大帮助。

Parallel.ForEach(namelist, /*new ParallelOptions { MaxDegreeOfParallelism = 4 }, */line =>
//foreach (string line in namelist)
{
      if (worker.CancellationPending)
      {
             e.Cancel = true;
             worker.ReportProgress(SysCount, obj);
      }
      else
      {
             this.SystemName = line;//file.ReadLine();
             Status.sVariables result = new Status.sVariables();
             result = OneSystem(this.SystemName);
             switch (result.BGWResult)
             {
                    case -1:
                       this.StatusString = "Login to server failed!";
                       break;
                    //other status are assigned here;
             }
             SysCount++;
             bag.Add(this);
      }
      Status returnobj;
      bag.TryTake(out returnobj);
      worker.ReportProgress(SysCount, returnobj);
      Thread.Sleep(200);
});

ReportProgress方法：

private void backgroundWorker1_ProgressChanged(object sender, ProgressChangedEventArgs e)
    {
        if (!backgroundWorker1.CancellationPending)
        {
            Status result = (Status)e.UserState;
            Complete_label.Visible = true;
            if (listView1.InvokeRequired)
                listView1.Invoke(new MethodInvoker(delegate
                {
                    listView1.Items.Add("");
                    listView1.Items[result.SysCount - 1].SubItems.Add(result.SystemName);
                    listView1.Items[result.SysCount - 1].SubItems.Add(result.StatusString);
                    listView1.Items[result.SysCount - 1].SubItems.Add(result.AvailableUpdatesCount.ToString());

                }));
            else
            {
                try
                {
                    listView1.Items.Add("");
                    listView1.Items[result.SysCount - 1].SubItems.Add(result.SystemName);
                    listView1.Items[result.SysCount - 1].SubItems.Add(result.StatusString);
                    listView1.Items[result.SysCount - 1].SubItems.Add(result.AvailableUpdatesCount.ToString());
                }
                catch (Exception ex)
                {}
                //other stuff
           }
    }

Answer 1

真正的问题是ListView更新代码使用错误的索引来更新项。它假定Status.SysCount属性包含正确的索引。如果执行按顺序进行，这可能是正确的，但如果执行以并行方式运行，则会失败--不同的线程可以以不同的速度完成，并报告进度无序。

通过使用ListViewItem返回的ListViewItemCollection.Add对象，可以简单地解决实际问题。

private void backgroundWorker1_ProgressChanged(object sender, ProgressChangedEventArgs e)
{
    if (!backgroundWorker1.CancellationPending)
    {
        Status result = (Status)e.UserState;
        Complete_label.Visible = true;
        var newItem=listView1.Items.Add("");
        newItem.SubItems.Add(result.SystemName);
        newItem.SubItems.Add(result.StatusString);
        newItem.SubItems.Add(result.AvailableUpdatesCount.ToString());
        //other stuff
    }
}    

但是，代码有更严重的问题-- State类尝试并行处理数据，将数据存储在自己的属性中，然后将其发送给报表。显然，显示的数据总是在变化。

一个更好的选择是在循环中创建一个新的State实例，或者更好的是创建一个仅用于报告的类：

class StatusProgress 
{
    public string SystemName{get;set;}
    public string StatusString{get;set;}
    public int AvailableUpdatesCount {get;set;}
}

....
int sysCount=0;
Parallel.ForEach(namelist, line =>
{
    var progress=new StatusProgress();
    progress.SystemName = line;//file.ReadLine();

    Status.sVariables result = new Status.sVariables();
    result = OneSystem(line);
    switch (result.BGWResult)
    {
        case -1:
            progress.StatusString = "Login to server failed!";
            break;
                //other status are assigned here;
    }
    var count=Interlocked.Increment(ref sysCount);
  }
  worker.ReportProgress(count, progress);
});

请注意，使用SysCount++代替Interlocked.Increment来原子地增加值，并获得增量值的副本。如果没有这样做，多个线程可以在我有机会报告进度之前修改SysCount。

进度报告代码将更改为使用StateProgress

        StatusProgress result = (StatusProgress)e.UserState;

最后，BackgroundWorker已经过时，因为任务并行库以更加轻量级的方式提供了BGW所做的一切以及更多的功能。例如，您可以使用取消并行循环使用CancellationToken并使用进展类以类型安全的方式报告进度。

.NET中的大多数异步方法都识别CancellationToken和Progress，这意味着您可以轻松地以在这里显示的形式报告进度和取消异步任务。

代码可以这样重写：

在UI表单上：

private void ReportServerProgress(StatusProgress result)
{
    Complete_label.Visible = true;
    var newItem=listView1.Items.Add("");
    newItem.SubItems.Add(result.SystemName);
    newItem.SubItems.Add(result.StatusString);
    newItem.SubItems.Add(result.AvailableUpdatesCount.ToString());
    //other stuff
}   

CancellationTokenSource _cts;
Progress<StatusProgress> _progress;

public void StartProcessiong()
{
    _cts=new CancellationTokenSource();
   _progress=new Progress<StatusProgress(progress=>ReportServerProgress(progress);
   StartProcessing(/*input*/,_cts.Token,_progress);
}

public void CancelLoop()
{
   if (_cts!=null)
      _cts.Cancel();
}

处理代码可以在同一窗体上，也可以在任何其他类上。实际上，最好将UI与处理代码分开，特别是当您有非平凡的处理时，例如调用每个服务器来确定其状态。

public void StartProcessing(/*input parameters*/,
                    CancellationTokenSource token,
                    IProgress<StatusProgress> progress)
{
    .....
    var po=new ParallelOptions();
    po.CancellationToken=token;
    Parallel.ForEach(namelist, po,line =>
    {
       var status=new StatusProgress();
       status.SystemName = line;//file.ReadLine();
       Status.sVariables result = new Status.sVariables();
       result = OneSystem(line);
       switch (result.BGWResult)
       {
           case -1:
               progress.StatusString = "Login to server failed!";
               break;
               //other status are assigned here;
       }
       progress.Report(status);
    }
}

许多异步.NET方法都接受取消令牌，因此可以将它传递给Web调用，并确保循环和任何未完成的长调用都被取消。

Answer 2

您的结果都混淆了，因为您使用并行操作写入全局状态，例如SystemName和StatusString，因此当您试图读取和打印它们的值时，这些全局变量的内容将全部混淆。

您可以引入一个lock，但这将完全消除Parallel.ForEach的缺点。因此，要么放弃使用Parallel.ForEach (在这种情况下这似乎没有什么用处)，要么您需要收集数据并确保以线程安全的方式将其发送给记者。

为了进一步解释，让我们检查一下代码：

this.SystemName = line; // <- the worker has now written to this, which is global to all workers
...
result = OneSystem(this.SystemName); // <- another worker may have overwritten SystemName at this point
...
                   this.StatusString = "Login to server failed!"; // <- again writing to shared variable
         ...
         bag.Add(this); // <- now trying to "thread protect" already corrupted data

因此，如果必须并行运行循环，每个工作人员必须只更新自己的孤立数据，然后将其推送到GUI编组报告方法。