Parallel.For 是否支持写入数组？

Question

这是我在 MSDN 网站上找到的一些代码。

long[] nums = { 1, 2, 3, 4 };
long total = 0;
Parallel.For<long>(0, nums.Length, () => 0, (j, loop, subtotal) =>
{
    subtotal += nums[j];
    return subtotal;
}, (x) => Interlocked.Add(ref total, x));

我想知道是否有人可以告诉我如何修改此代码以将结果写回共享数组？如下所示。

My_Class[] localArray;
Parallel.For<My_Class[]>(0, localArray.Length, () => 0, (j, loop, messages)  =>
{
    List<My_Class> tempList = new List<My_Class>();
    My_Class myClass = localArray[j];
    foreach (Message msg in myClass.GetListOfMessages())
    {
         if (msg.getHeader() == "Hello")
         {
              tempList.add(msg);
         }
    }
    messages = temp.toArray();
    return messages;
}, (x) => Interlocked.Add(ref localArray, x));

所以你看到上面的例子我有一个共享的本地数组，其中包含“My_Class”类型的对象。 My_Class 对象包含一个消息列表，我要搜索这些消息以查找标题为“Hello”的特定类型的消息。

因此，在示例代码中，我使用并行 for 循环遍历我的自定义对象并创建一个名为“tempList”的临时列表，我最终将其转换为数组并作为结果返回。

但是上面的代码只是不适用于数组，特别是 (x) => Interlocked 行，它似乎不兼容将许多数组返回的结果合并到一个数组中。

请有人帮助我。

Answer 1

联锁只支持两种添加方法：

Add(Int32, Int32) 
// Adds two 32-bit integers and replaces the first integer with the sum, as an atomic operation.

Add(Int64, Int64)   
// Adds two 64-bit integers and replaces the first integer with the sum, as an atomic operation.

所以我认为它不适用于某种对象。

Answer 2

你试图使用不恰当的例子。

MSDN 代码演示了对大量数据进行数值处理以通过对数据进行分区并生成小计，然后将小计收集到最终结果中来生成单个结果。第二步由 Interlocked.Add 执行，因为小计的数量会非常少。

您的代码做了一些完全不同的事情，过滤/处理大量消息而不尝试聚合部分结果。

查找消息的最简单方法是使用 PLINQ 和 Where：

var helloMessages=from msgClass in  localArray.AsParallel()
                  from msg in msgClass.GetListOfMessages()
                  where msg.getHeader()=="Hello"
                  select msg;
var results=helloMessages.ToArray();

与纯 LINQ 查询的唯一区别是 AsParallel() 关键字。同样可以直接使用扩展方法来写：

var fesults=localArray.AsParallel()
                      .SelectMany(mscClass=>msgClass.GetListOfMessages())
                      .Where(msg=>msg.getHeader()=="Hello")
                      .ToArray();

如果只是为了使代码更简洁，您可能应该使用属性而不是 getter 方法。

var fesults=localArray.AsParallel()
                      .SelectMany(mscClass=>msgClass.ListOfMessages)
                      .Where(msg=>msg.Header=="Hello")
                      .ToArray();

如果出于某种原因您想将数据写入集合，您应该使用并发集合之一，例如ConcurrentQueue，例如ConcurrentQueue<Message>：

ConcurrentQueue<Message> results=new ConcurrentQueue<Message>();
Parallel.ForEach(longMessageCollection,msg=>
        {
            //Do some really complex processing to select the message
            results.Enqueue(msg);
        });