并行处理混淆

Question

我是C＃编程新手。 我想获取使用后台工作服务器列表的更新数量。每个服务器的结果都以列表视图的报告进度方式显示。 我能够使用foreach循环成功获取结果，但在尝试使用并行foreach获得相同结果时，列表视图的所有列和行都会混淆在一起。

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

1. server1的登录服务器失败！ 0
2. 服务器2更新可用3个
3. 服务器3更新可用3
4. 服务器4更新0 等..并行的foreach

输出：

1. server1的更新可用1
2. server1登录到服务器失败！ 1
3. server2登录服务器失败！ 0
4. server3登录服务器失败！ 0
5. server4登录服务器失败！ 0
6. 服务器4更新可用3 等..

我曾尝试锁定部分代码并且也使用并发袋尝试，但不太能解决这个问题。以下是parallelforeach代码。我做错了什么？任何建议都会有很大的帮助。

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类尝试并行处理数据，存储在其自己的数据属性，然后发送自己的报告。显然，所显示的数据将始终在变化。

更好的选择是要么建立在循环中一个新的国家的实例，或者更好的，只为报告创建一个类：

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来cancel the parallel loop，并使用Progress类以类型安全的方式报告进度。

.NET中的大多数异步方法都可以识别CancellationToken和Progress，这意味着您可以轻松地报告进度并取消异步任务，如shown here。

该代码可以被重写如下：

在一个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编组报告方法。