多线程的多次迭代

Question

我重写了最初只针对一个线程的方法，以便与多个线程一起工作。现在，此方法接受两个并发集合：ConcurrentBag，即List，和ConcurrentQueue，即Queue。

目的是匹配两个集合中的两个标题并进行一些逻辑，这是ConcurrentBag项目中的简单赋值。我当然知道ConcurrentBag 中的所有符号都在ConcurrentQueue 中。

当我为多线程编写此内容时，发生了一些标题不匹配 (~20%)，这在线程上没有发生。只有在调试期间，我才能匹配这些标题，然后分配值。迭代这两个集合一定有一些问题。也许在同一时间许多线程从同一个项目中读取值，但只读取应该没有问题？

以下代码：

public void UpdateWithPercent(ref ConcurrentBag<Book> refList, ConcurrentQueue<Book> list)
{
    var size = list.Count;
    int numProcs = Environment.ProcessorCount;
    var divider = CalculatBiggestDivider(size);
    var nextIteration = 0;
    var remainingWork = numProcs;
    var internalRefList = refList;
    using (ManualResetEvent mre = new ManualResetEvent(false))
    {
        for (int i = 0; i < numProcs; i++)
        {
            ThreadPool.QueueUserWorkItem(delegate
            {
                IEnumerable<Book> partialList;
                while (-(nextIteration - Interlocked.Add(ref nextIteration, (partialList = DequeueChunk(list, divider)).Count()))> 0)
                {
                    foreach (var item in partialList)
                    {
                        foreach (var x in internalRefList)
                        {
                            if (x.Title == item.Title)
                            {
                                x.Orders += item.Orders;
                                break;
                            }
                        };
                    }
                }

                if (Interlocked.Decrement(ref remainingWork) == 0)
                {
                    mre.Set();
                }
            });
        }

        mre.WaitOne();
    }

    refList = internalRefList;
}

private int CalculatBiggestDivider(int count)
{
    var divider = 1;
    for (int i = 30; i > 0; i--)
    {
        if (count % i == 0)
        {
            divider = i;
            break;
        }
    }

    return divider;
}

private IEnumerable<T> DequeueChunk<T>(ConcurrentQueue<T> queue, int chunkSize)
{
    for (int i = 0; i < chunkSize && queue.Count > 0; i++)
    {
        T item;
        bool success = queue.TryDequeue(out item);
        if (!success)
        {
            i = chunkSize;
            continue;
        }
        yield return item;
    }
}

Answer 1

最终我决定退出嵌套循环并使用ConcurrentDictionary。现在可以了。

public void UpdateWithPercent(ref ConcurrentDictionary<string, Book> refList, List<Book> list, int ticker, int maxTimes)
{
    var size = list.Count;
    int numProcs = Environment.ProcessorCount;
    var divider = CalculatBiggestDivider(size);
    var nextIteration = 0;
    var remainingWork = numProcs;
    var internalRefList = refList;
    using (ManualResetEvent mre = new ManualResetEvent(false))
    {
        for (int i = 0; i < numProcs; i++)
        {
            ThreadPool.QueueUserWorkItem(delegate
            {
                int index = 0;
                while ((index = Interlocked.Add(ref nextIteration, divider) - divider) < size)
                {
                    foreach (var item in list.GetRange(index, divider))
                    {
                        Book x;
                        if (internalRefList.TryGetValue(item.Title, out x))
                        {
                                x.Orders += item.Orders;
                        }
                    };
                }

                if (Interlocked.Decrement(ref remainingWork) == 0)
                {
                    mre.Set();
                }
            });
        }

        mre.WaitOne();
    }

    refList = internalRefList;
}