多线程不处理所有任务

Question

我正在使用以下代码使用线程来执行任务，这里我尝试执行“dtTable”数据表中的所有记录。有限的线程数为 2（即有时只允许两个线程/只允许执行）。问题是它没有执行 Datatable 中可用的所有记录，它以不规则的方式执行数据。可能是什么问题..？提前致谢。

 public class Generator : IDisposable
 {
    public static int maxThreadCount = 2;
    public static int runningThreadCount = 0;

    public RunTask()
    {
       for (int ro = 0; ro <= dtTable.Rows.Count - 1; ro++)
       {
          if (maxThreadCount > runningThreadCount)
          {
            Thread atpthread = new Thread(delegate()
            {
            DoOperationMethod(dtTable.Rows[ro], Task, startDate, EndDate, dtTemplate);
            });
            atpthread.Start();
            runningThreadCount = runningThreadCount + 1;
            Mainthreads.Add(atpthread);
          }
          else
          {
            ro--;
          }
       }
  }

  public void DoOperationMethod(DataRow drAttachpoint, System.StrTaskItem Task, DateTime startDate, DateTime EndDate, DataTable dtTemplate)
    {
     //doing my Operation
     runningThreadCount = runningThreadCount-1; //Once Task done count will get reduce
    }

}

我正在使用 .net 3.5（仅供参考）。

Answer 1

问题是您在等待线程空闲时不断迭代数据表中的行。无论如何，这种多线程，即使它实际上是正确的（它不是），也是相当低效的——大部分时间可能都花在启动新线程上。

试试这样的：

Parallel
 .ForEach(dtTable.Rows.OfType<DataRow>(), row => DoOperationMethod(...))
 .WithDegreeOfParallelism(2);

编辑：

要弄清楚问题是如何产生的，您必须了解如何在匿名方法中捕获变量。您的DoOperationMethod 调用没有被传递到您想要的数据行，因为ro“变量”没有被复制，而是被引用。因此，当ro 在循环中发生变化时，它也会在您创建的线程中发生变化。

除了你的代码是可怕的线程不安全和低效的事实之外：

• 实际上，您正在浪费 三个 线程来工作 - 没有阻塞，您的循环只是不断地加减 ro，这几乎完全是纯 CPU 工作。这比在等待结果时简单地阻塞更浪费。
• 您不能只从多个线程读取和写入静态字段并期望事情正常工作。实际上，您的代码很容易并行启动比您想要的更多的线程 - 甚至在没有线程运行的情况下，runningThread 最终成为2 使整个事情陷入僵局。
• 您不断启动新线程来执行看似微不足道的操作 - 我猜您的大部分工作要么受 I/O 限制，要么受制于一遍又一遍地创建新线程的成本。
• 我假设 Mainthreads 是某种类型的列表，并且我假设您也从 DoOperationMethod 方法修改它 - 同样，这可能会导致随机异常和意外结果。
• 理论上，由于各种优化和缓存，检查maxThreadCount > runningThreadCount 甚至可能永远不会被评估。实际上，在当前 x86 CPU 上的 .NET 上，这种复杂的方法不太可能出现这种情况，但是当您更新到 .NET 7.0 或其他版本时，这种情况可能会咬到你 :)

多线程是困难的。你真的不想猜测你的方式。至少，尝试先了解基础知识 - http://www.albahari.com/threading/。

Answer 2

在我看来，您的代码最大的问题是，即使您修复了 runningThreadCount 的线程不安全用法，您的代码在等待某个线程完成时也会“旋转”。当您尝试完成实际工作时，这会完全占用 CPU 内核。

Luann 提出的解决方案很好，尽管我会使用Cast<DataRow>() 而不是OfType<DataRow>()（因为枚举中的所有元素实际上都应该是DataRow 类型）。除了简洁之外，它的一大优点是它使用了线程池，这将大大减少线程管理的开销（因为它重用线程而不是一遍又一遍地创建和销毁它们）。

如果您更喜欢更明确的方法，您可以修改您发布的代码以使用信号量：

SemaphoreSlim semaphore = new SemaphoreSlim(2);

for (int ro = 0; ro <= dtTable.Rows.Count - 1; ro++)
{
    semaphore.Wait();

    DataRow row = dtTable.Rows[ro];

    Thread atpthread = new Thread(delegate()
    {
        DoOperationMethod(row, Task, startDate, EndDate, dtTemplate);
        semaphore.Release();
    });
    atpthread.Start();
    Mainthreads.Add(atpthread);
}

这将导致主线程在信号量计数达到 0 时阻塞 Wait() 调用，并在计数再次为正时（即在线程调用 Release() 之后）继续。

我注意到评论者 Enigmativity 关于dtTable 是否安全使用的观点。我在这里假设对象在此处理过程中没有被修改。有了这个假设，不同步使用它应该没问题。但如果这个假设是错误的，那么我同意他们的观点，那就是代码中的另一个错误。

最后，我要指出，您看到行被跳过的原因是您在匿名方法中使用了变量ro。在匿名方法开始执行循环的给定迭代之前，该变量可以很容易地递增，从而导致该线程处理错误的行。某些行可能会被处理多次，而其他行则被跳过。我已经通过在循环块内的变量中检索DataRow 对象来解决上述代码示例中的问题，以便每个线程都获得它自己的变量的私有副本，以及它应该处理的DataRow 对象.