在 ConcurrentDictionary.GetOrAdd 中使用 lock 语句时的已知副作用？

Question

从ConcurrentDictionary ValueFactory 方法内部的多个线程添加到实体框架DbSet 时遇到了一些问题。我试图通过引入lock 声明来消除这个问题。不过，这似乎有一些奇怪的副作用。在一些罕见和随机的情况下，我的代码会抛出 KeyNotFoundException，即使编程应该防止这种情况发生。我想我在监督一些事情。

using (ESBClient client = new ESBClient()) { // WCF SERVICE

    client.Open();

    // Limit the maximum number of parallel requests
    var esbLimiter = new SemaphoreSlim(4);

    ConcurrentDictionary<string, DataEntry> dataEntryDict  = new ConcurrentDictionary<string, DataEntry>(
        await db.DataEntries
            .Where(de => allObjIDs.Contains(de.PAObjID))
            .IncludeOptimized(de => de.WorkSchedules)
            .ToDictionaryAsync(a => a.PAObjID, a => a)
    );


    // Get WorkOrderDataSet02 for each data entry number
    await Task.WhenAll(allDataEntryNumbers.Batch(20).Select(async workOrderBatch => {
        await esbLimiter.WaitAsync();

        Debug.WriteLine($"Starting for new batch after {s.ElapsedMilliseconds} with parallel {esbLimiter.CurrentCount}");

        try {
            int retryCounter = 0;
            getWorkOrderDataSet02Response gwoResp;

            retryCurrentWorkOrderDataSetResp:
            try {
                gwoResp = await client.getWorkOrderDataSet02Async(
                    new getWorkOrderDataSet02Request(
                        "?",
                        companyGroup.Key,
                        string.Join(",", workOrderBatch.Select(wob => wob.DataEntryNumber)),
                        "WNTREIB",
                        "?",
                        "act,sales",
                        "D"
                    )
                );

            } catch (System.ServiceModel.CommunicationException ex) {
                // Retry up to 3 times before finally crashing
                if (retryCounter++ < 3) {
                    await HandleServiceRetryError("getWorkOrderDataSet02Async", retryCounter, s.ElapsedMilliseconds, ex);
                    goto retryCurrentWorkOrderDataSetResp;
                } else
                    throw;
            }

            // Iterate over all work orders returned by the ESB
            foreach (dsyWorkOrder01TtyWorkOrder currDetail in gwoResp.dsyWorkOrder01) { // dsyWorkOrder01 IS AN ARRAY OF OBJECTS. IT COMES FROM A WCF CALL. PAObjID IS UNIQUE.
                // Get or create element
                DataEntry currentEntry = dataEntryDict.GetOrAdd(
                    currDetail.Obj,
                    key => {
                        DataEntry newDe = new DataEntry();
                        lock (db.DataEntries) { // I INTRODUCED THOSE LOCK STATEMENTS
                            db.DataEntries.Add(newDe); // THIS IS THE LINE THAT WAS PROBLEMATIC IN THE FIRST PLACE
                        }
                        return newDe;
                    }
                );

                // Set regular fields
                currentEntry.ApplyTtyWorkOrder(currDetail, resourceDict); // THIS METHOD APPLIES THE PAObjID PROPERTY
            }

            // Delete all elements, that were not provided by the service anymore
            lock(db.DataEntries) { 
                workOrderBatch
                    .Where(wob => !gwoResp.dsyWorkOrder01
                        .Where(wo => wo.DataEntryNumber.HasValue)
                        .Select(wo => wo.DataEntryNumber.Value)
                        .Contains(wob.DataEntryNumber)
                    )
                    .ToArray()
                    .ForEach(dataEntry => {
                        try {
                            db.DataEntries.Remove(dataEntryDict[dataEntry.ObjID]); // THIS LINE THROWS THE KeyNotFoundException
                        } catch (Exception ex) {
                            throw new Exception($"Key {dataEntry.ObjID} not in list.", ex);
                        }

                    });
            }

            // Update progress
            progress.Report(.1f + totalSteps * Interlocked.Increment(ref currentStep) * .8f);

        } finally {
            Debug.WriteLine($"Finished for batch after {s.ElapsedMilliseconds} with parallel {esbLimiter.CurrentCount}");
            esbLimiter.Release();
        }

    }));
}

// HERE'S THE APPLY METHOD
public void ApplyTtyWorkOrder(dsyWorkOrder01TtyWorkOrder src, Dictionary<(string Name, byte ResourceType), int> resourceDict) {
    Deleted = false;

    DataEntryNumber = src.DataEntryNumber.Value;
    PAObjID = src.Obj; // PAObjID IS APPLIED HERE
    IsHeader = src.IsHeader;
    Pieces = Convert.ToInt16(src.ProductionQty);
    PartNo = src.Article;
    JobNo = src.WorkOrder;
    StartDate = src.StartDate;
    FinishDate = src.EndDate;
    FinishedPA = src.WorkOrderStatus == "R";

    // Update methods
    UpdateFromTtyCustomer(src.ttyCustomer?.FirstOrDefault());
    UpdateFromPart(src.ttyPart?.FirstOrDefault());
    UpdateFromSalesDocHeader(src.ttySalesDocHeader?.FirstOrDefault());
    UpdateWorkSchedules(src.ttyWorkOrderActivity, resourceDict);
}

我在我认为相关的每一行都添加了一个大写注释。

我不知道为什么会发生这个错误。据我了解，我只尝试从 dataEntryDict 字典 dataEntry.ObjID 键中获取一个条目，这些键是我之前在循环的同一迭代中添加的。

在我介绍这两个锁语句之前，标有“这是第一个地方有问题的行”的行偶尔会抛出异常：“集合已修改；枚举操作可能无法执行。”深挖EF的代码后，我意识到这应该与DbSet.Add方法的实现方式有关。

在ValueFactory 中使用lock 语句时是否有任何已知的副作用？

Answer 1

lock (db.DataEntries) { // I INTRODUCED THOSE LOCK STATEMENTS
    db.DataEntries.Add(newDe); // THIS IS THE LINE THAT WAS PROBLEMATIC IN THE FIRST PLACE
}

问题在于db.DataEntries 不是线程安全的集合，而是被多个线程同时访问。所有 EF 对象都不是线程安全的。

在这里使用锁定似乎是一个很好的解决方案。确保你抓住了所有的地方。

通常最好将并发部分与顺序部分分开。 only 使client.getWorkOrderDataSet02Async 调用并发并将结果收集到一个集合中。然后，依次处理结果。