在 .NET 中使用 Interlocked 类进行多线程处理的正确方法

Question

我有一个计数器，用来统计当前处理的大报表

private int processedLargeReports;

我正在生成并启动五个线程，每个线程都访问此方法：

public bool GenerateReport(EstimatedReportSize reportSize)
{
    var currentDateTime = DateTimeFactory.Instance.DateTimeNow;
    bool allowLargeReports = (this.processedLargeReports < Settings.Default.LargeReportLimit);
    var reportOrderNextInQueue = this.ReportOrderLogic.GetNextReportOrderAndLock(
        currentDateTime.AddHours(
        this.timeoutValueInHoursBeforeReleaseLock), 
        reportSize, 
        CorrelationIdForPickingReport, 
        allowLargeReports);

    if (reportOrderNextInQueue.IsProcessing)
    {
        Interlocked.Increment(ref this.processedLargeReports);                
    }

    var currentReport = this.GetReportToBeWorked(reportOrderNextInQueue);

    var works = this.WorkTheReport(reportOrderNextInQueue, currentReport, currentDateTime);
    if (reportOrderNextInQueue.IsProcessing)
    {
        Interlocked.Decrement(ref this.processedLargeReports);                
    }
    return works;           
}

“reportOrderNextInQueue”变量从数据库中获取报告顺序，并检查报告顺序是“正常”还是“大”（这是通过定义 reportOrderNextInQueue 变量的 bool IsProcessing 属性来实现的）。在大报表的情况下，系统然后 Interlock 递增处理的大报表 int 并处理大报表。处理完大型报表后，系统联锁会递减该值。

整个想法是我一次只允许处理一个报告，因此一旦一个线程正在处理一个大报告，其他线程应该无法访问数据库中的一个大报告。 bool allowLargeReport 变量检查handledLargeReports int 和是否超过限制。

我很好奇这是否是正确的实现，因为我无法在星期一之前对其进行测试。我不确定我是否必须使用 InterLocked 类或只是将 processesLargeReports 变量定义为 volatile 成员。

Answer 1

假设您有 5 个线程开始运行上面的代码，LargeReportLimit 为 1。它们都将 processesLargeReports 读取为 0，allowLargeReports 对它们为真，并且它们将同时开始处理 5 个项目，尽管您的限制为 1。所以如果我理解正确的话，我真的看不出这段代码是如何实现你的目标的。

稍微扩展一下：您阅读已处理的LargeReports，然后执行它（使用它来检查您是否应该允许处理报告）。你表现得好像这个变量不能在读取和行为之间改变，但事实并非如此。任何数量的线程都可以在你读取和作用于变量之间对处理的大报告做任何事情，因为你没有锁定。在这种情况下，Interlocked 只会确保在所有线程处理完所有任务后，processedLargeReports 将始终为 0，但仅此而已。

如果您需要限制对某些资源的并发访问 - 只需为此使用适当的工具：Semaphore 或 SemaphoreSlim 类。创建允许 LargeReportLimit 线程进入的信号量。在处理报告之前，请等待您的信号量。如果达到并发线程处理报告的数量，这将阻塞。处理完成后，释放信号量以允许等待线程进入。此处无需使用 Interlocked 类。

Answer 2

volatile 不提供线程安全。与多线程一样，您需要一些同步 - 它可以基于 Interlocked、lock 或任何其他同步原语，具体取决于您的需要。你选择了Interlocked - 很好，但是你有一个竞争条件。您在任何同步块之外读取processedLargeReports 字段并根据该值做出决定。但它可能会在您阅读后立即更改，因此整个逻辑将不起作用。正确的方法是始终执行 Interlocked.Increment 并将您的逻辑基于返回的值。像这样的：

首先，让该字段使用更好的名称

private int processingLargeReports;

然后

public bool GenerateReport(EstimatedReportSize reportSize)
{
    var currentDateTime = DateTimeFactory.Instance.DateTimeNow;
    bool allowLargeReports = 
       (Interlocked.Increment(ref this.processingLargeReports) <= Settings.Default.LargeReportLimit);
    if (!allowLargeReports)
        Interlocked.Decrement(ref this.processingLargeReports);
    var reportOrderNextInQueue = this.ReportOrderLogic.GetNextReportOrderAndLock(
        currentDateTime.AddHours(
        this.timeoutValueInHoursBeforeReleaseLock), 
        reportSize, 
        CorrelationIdForPickingReport, 
        allowLargeReports);
    if (allowLargeReports && !reportOrderNextInQueue.IsProcessing)
        Interlocked.Decrement(ref this.processingLargeReports);

    var currentReport = this.GetReportToBeWorked(reportOrderNextInQueue);

    var works = this.WorkTheReport(reportOrderNextInQueue, currentReport, currentDateTime);
    if (allowLargeReports && reportOrderNextInQueue.IsProcessing)
        Interlocked.Decrement(ref this.processingLargeReports);
    return works;           
}

请注意，这也包含竞争条件，但包含您的 LargeReportLimit 约束。

编辑：现在我在想，由于您的处理是基于 Allow 和 Is 大型报告，Interlocked 是不是一个好的选择，最好使用基于Monitor 的方法，例如：

private int processingLargeReports;
private object processingLargeReportsLock = new object();

private void AcquireProcessingLargeReportsLock(ref bool lockTaken)
{
    Monitor.Enter(this.processingLargeReportsLock, ref lockTaken); 
}

private void ReleaseProcessingLargeReportsLock(ref bool lockTaken)
{
    if (!lockTaken) return;
    Monitor.Exit(this.processingLargeReportsLock);
    lockTaken = false;
}

public bool GenerateReport(EstimatedReportSize reportSize)
{
    bool lockTaken = false;
    try
    {
        this.AcquireProcessingLargeReportsLock(ref lockTaken); 
        bool allowLargeReports = (this.processingLargeReports < Settings.Default.LargeReportLimit);
        if (!allowLargeReports)
        {
            this.ReleaseProcessingLargeReportsLock(ref lockTaken);
        }
        var currentDateTime = DateTimeFactory.Instance.DateTimeNow;
        var reportOrderNextInQueue = this.ReportOrderLogic.GetNextReportOrderAndLock(
            currentDateTime.AddHours(
            this.timeoutValueInHoursBeforeReleaseLock), 
            reportSize, 
            CorrelationIdForPickingReport, 
            allowLargeReports);
        if (reportOrderNextInQueue.IsProcessing)
        {
            this.processingLargeReports++;
            this.ReleaseProcessingLargeReportsLock(ref lockTaken);
        }            
        var currentReport = this.GetReportToBeWorked(reportOrderNextInQueue);
        var works = this.WorkTheReport(reportOrderNextInQueue, currentReport, currentDateTime);
        if (reportOrderNextInQueue.IsProcessing)
        {
            this.AcquireProcessingLargeReportsLock(ref lockTaken); 
            this.processingLargeReports--;
        }            
        return works;
    }
    finally
    {
        this.ReleaseProcessingLargeReportsLock(ref lockTaken);
    }           
}