在.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”变量从数据库中获取报告订单，并检查报告订单是否为“Normal”或“Large”（这是通过定义reportOrderNextInQueue变量的bool IsProcessing属性实现的）。在处理大型报告时，系统会Interlock增加processedLargeReport int并处理大型报告。一旦大型报告被处理，系统会Interlock减少该值。

整个想法是只允许同时处理一个报告，因此一旦一个线程正在处理大型报告，则其他线程不应能够访问数据库中的大型报告。bool allowLargeReport变量检查processedLargeReports int是否超过限制。

我很好奇这是否是正确的实现方式，因为在周一之前我无法进行测试。我不确定是否必须使用InterLocked类还是只需将processedLargeReports变量定义为volatile成员。

- Osman Esen

---

你可能需要一个“锁”而不是一个增量。作为学术练习，可以参考Eric Lippert的文章《C#中的锁定机制》来深入了解它的一般工作原理。 - theB

@theB提供的链接已经失效。原始博客可以在https://ericlippert.com/2014/02/12/how-does-a-lock-work/找到，但主要只是指向同一原始网站的损坏链接。 - shapeshifter42

Answer 1

假设你有5个线程开始运行以上代码，并且LargeReportLimit为1。它们都会读取processedLargeReports的值为0，allowLargeReports也是true，因此它们将同时开始处理5个项目，尽管您的限制是1。所以如果我正确理解的话，我并不认为这段代码能够实现你的目标。

再具体地说明一下：你先是读取了processedLargeReports，然后根据它采取相应的行动(使用它来检查是否应允许处理报告)。你却像这个变量在读取和采取行动之间不能被改变，但事实并非如此。任何数量的线程都可以在你读取和采取行动之间对processedLargeReports进行任何操作，因为你没有锁定。在这种情况下，Interlocked仅确保在所有线程完成处理所有任务后，processedLargeReports始终会变成0，但这就是全部的效果了。

如果需要限制对某些资源的并发访问，请使用适当的工具：Semaphore或SemaphoreSlim类。创建一个允许LargeReportLimit个线程的信号量。在处理报告之前，在信号量上等待。如果达到了处理报告的并发线程数，则此操作将阻止。当处理完成时，释放信号量以允许等待的线程进入。这里不需要使用Interlocked类。

- Evk

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约束。

编辑：现在我在考虑，由于您的处理基于“允许”和“是”大型报告，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);
    }           
}

- Ivan Stoev

---

嗯，C#的lock构造只是对Monitor的一种语法糖，因此使用它们中的任何一个都被认为是一种锁定方法。在这种特殊情况下，我们需要能够在某些情况下提前释放锁（！allowLargeReports分支），同时在其他情况下保持它并稍后释放它。lock语句不提供这样的灵活性（实际上不允许），因此我使用了直接的方法。 - Ivan Stoev

严格来说，我只需要针对“增量”部分使用它，并且可以在“减量”部分使用lock，但由于我已经需要一个finally块，所以这样做没有任何好处。 - Ivan Stoev

似乎不同的任务没有共享 processingLargeReports 整型变量。即使 Task 1 把它增加到了 1，当 Task 2 访问该方法时，它仍然将该值视为 0。 - Osman Esen

我正在使用任务而不是线程。这会有任何区别吗？ - Osman Esen

@OsmanEsen：只要他们使用包含上面代码的类的相同实例，那就没关系。您确定没有在您的类的不同实例上调用"GenerateReport"方法吗？ - Ivan Stoev