带有 Parallel.ForEach 的线程安全代码

Question

我很少在我的代码中使用线程，但面临着减少运行时间的巨大压力，因此尝试利用 parallel.ForEach 循环。

代码对出价（项目）执行计算。这些项目已加载到List<List<Item>> 中（外部列表是所有相同 ID 的项目，内部列表是各种项目）。我在最后包含了类 Item。

List<List<Item>> queryItemsByAcceptID = phyBidList.GroupBy(bids => bids.acceptID)
                        .Select(group => group.ToList())
                        .ToList();

所有出价都有一个按时间分隔的属性（结算期），例如时段 1 = 00:00-00:30，时段 2 = 00:30-01:00，依此类推。 第 2 期的出价不需要来自第 1 期的出价的任何信息。 因此，我将投标分为结算期，在 parallel.ForEach 中运行计算。

List<Item> phyBidList = new List<Item>();
var queryMassBySetPeriod = phyBidList.GroupBy(x => x.settlementPeriod)
                                .Select(group => group.ToList())
                                .ToList();

但是，当我运行代码时，在多次运行时，“unParallel 代码”和以前的输出都会得到不一致的结果。这让我认为我的代码不是“线程安全的”，因为“非并行代码”既正确又一致（但太慢了）。

这个线程安全吗？我应该怎么做才能产生一致的结果？

我想知道我是否应该把日光锁在外面......

acceptIdItem.FPN.Add(fpn);
acceptIdItem.qAboPosArea.Add(tempQABOposArea);
acceptIdItem.qAboNegArea.Add(tempQABOnegArea);

但是，我不确定锁定是否合适，因为线程不（或至少不应该）访问相同的变量......两者都是因为不需要来自其他块的信息，并且只进行出价通过一次计算。

附：我在下面包含了代码，我尝试删除我认为不必要的代码，以使其更短且更易于阅读。

Parallel.ForEach(queryMassBySetPeriod, block =>
{

Console.WriteLine("GroupBy UnitID");    
var queryItemsByUnitID = block.GroupBy(bids => bids.unitID)
                        .Select(group => group.ToList())
                        .ToList();

Console.WriteLine("GroupBy AcceptID");
queryItemsByAcceptID = block.GroupBy(bids => bids.acceptID)
                        .Select(group => group.ToList())
                        .ToList();

Console.WriteLine("Beginning mass interpretation...");
foreach (var list in queryItemsByAcceptID)
{    
    int bY = 0;
    foreach (var acceptIdItem in list)
    { 
        DateTime fromTime = acceptIdItem.fromTime;
        DateTime toTime = acceptIdItem.toTime;

        TimeSpan duration = toTime - fromTime;

        for (int i = 0; i < (duration.Minutes); i++) //qTime fix (duration.Minutes + 1)
        {  
            var queryPNdata = (from item in PNList
                            where item.unitID == acceptIdItem.unitID && item.fromTime <= fromTime && item.toTime >= fromTime
                            select item).FirstOrDefault();

            int time = (acceptIdItem.qTimes[i] - acceptIdItem.fromTime).Minutes;
            float boa = MathHelper.calcBOA(acceptIdItem.fromLevel, acceptIdItem.toLevel, (duration).Minutes, time);    
            float fpn = MathHelper.calcFPN(queryPNdata.fromLevel, queryPNdata.toLevel, duration.Minutes, time);

            acceptIdItem.qTimes.Add(fromTime + i * ((toTime - fromTime) / duration.Minutes));
            acceptIdItem.boa.Add(boa);
            acceptIdItem.FPN.Add(fpn);

            string[] tempBOUR = new string[6]; string[] tempBOLR = new string[6];
            float[] tempQABOneg = new float[6]; float[] tempQABOpos = new float[6];
            for (int k = 1; k < 7; k++)
            {
                //calculate tempBOUR/ tempBOLR/ tempQABOpos/ tempQABOneg
            }

            acceptIdItem.BOUR.Add(tempBOUR);
            acceptIdItem.BOLR.Add(tempBOLR);    
            acceptIdItem.qAboPos.Add(tempQABOpos);
            acceptIdItem.qAboNeg.Add(tempQABOneg);
        }

        int aZ = 0; //declared outside the loop to access later
        for (aZ = 0; aZ < (acceptIdItem.qAboNeg.Count() - 1); aZ++)
        {
            float[] tempQABOnegArea = new float[6]; float[] tempQABOposArea = new float[6];
            for (int k = 1; k < 7; k++)
            {
                //calculate tempQABOnegArea/ tempQABOposArea
            }

            acceptIdItem.qAboPosArea.Add(tempQABOposArea);
            acceptIdItem.qAboNegArea.Add(tempQABOnegArea);
        } 
        bY++;
    }
}
});

在开始时分配信息并将类添加到列表（phyBidList）中。这是课程...

class Item
{
    public String unitID, acceptID, prevAcceptID, type;
    public DateTime fromTime, toTime, acceptTime;
    public int settlementPeriod, duration;
    public List<DateTime> qTimes = new List<DateTime>();
    public List<string[]> BOLR = new List<string[]>();
    public List<string[]> BOUR = new List<string[]>();
    public List<float[]> qAboNeg = new List<float[]>();
    public List<float[]> qAboPos = new List<float[]>();
    public List<float> boa = new List<float>();
    public List<float> prevBOA = new List<float>();
    public List<float> FPN = new List<float>();
    public List<float[]> qAboNegArea = new List<float[]>();
    public List<float[]> qAboPosArea = new List<float[]>();
}

****编辑****

为了响应@calum-mcveigh，我使用下面的 sn-p 和其他相关更改将列表更改为 concurrentBags。但是，它仍然会产生不一致的结果。我已经把完整的代码放在这里https://pastebin.com/EgnE2285

ConcurrentBag<ConcurrentBag<Item>> queryMass = new ConcurrentBag<ConcurrentBag<Item>>();
foreach (var items in queryMassBySetPeriod)
{
    ConcurrentBag<Item> item = new ConcurrentBag<Item>();
    foreach (var bid in items)
    {                    
        item.Add(bid);
    }
    queryMass.Add(item);
}

Answer 1

为了确保线程安全，您可以使用并发集合，例如 ConcurrentBag<T> 而不是 List<T>

您可以阅读有关线程安全集合的更多信息here

Answer 2

变量queryItemsByAcceptID 在Parallel.ForEach 外部声明，但在其中设置和使用。已停止查找，但可能还有其他变量存在相同问题。