枚举器中的多线程答案

【问题标题】：Multi-threading in Enumerator枚举器中的多线程
【发布时间】：2018-02-02 08:24:39
【问题描述】：

我正在使用 C#，我有一个枚举器，我正在按顺序读取枚举器内的数据。

这是第三方库对象，不支持 Parallel.Foreach

while(enumerator.Next())
{
  var item = enumerator.Read();
  ProcessItem(item);
}

ProcessItem(Item item)
{
// Is lock required here
 if(item.prop == "somevalue")
   this._list.Add(item);
}

我想在阅读内容的同时在这里实现多线程。

while(enumerator.Next())
{
  // This code should run in a multi-threaded way

  var item = enumerator.Read();
  // ProcessItem method puts these items on a class level list property 
  // Is there any Lock required?
  ProcessItem(item);
}

我是多线程的新手。请分享任何满足上述要求的代码示例。

【问题讨论】：

参考这个codeproject.com/Articles/56575/Thread-safe-enumeration-in-C
你的意思是ProcessItem(item) 做了类似_list.Add(item) 的事情吗？根据_list 类型，它需要锁定。如果您扩展一些细节，它将有所帮助（显示 ProcessItem 的代码）
是的，根据某些条件，它会在列表中添加一些项目。例如：If(item.prop == "something") then _list.Add(item);
codeproject 中的示例可能不是您需要的。我不建议使用它，不仅因为它违反了宪兵DoNotUseLockedRegionOutsideMethodRule 规则（不仅是这条规则）。据我了解，我将使用带有锁定的IEnumerator 的自定义实现来获取对象。但这也取决于ProcessItems 的实际实现，如 Ashkan Nourzadeh 所述。随着请求更改其枚举中的列表，代码可能非常复杂，我想说，您应该搜索另一种方式，而不更改列表。
When to use a Parallel.ForEach loop instead of a regular foreach?的可能重复

标签： c# multithreading locking

【解决方案1】：

是的，需要一些锁定。您可以使用锁或使用并发集合类型来实现它。

使用lock:

ProcessItem(Item item)
{
    if(item.prop == "somevalue")
    {
        lock(_list)
        {
            _list.Add(item);
        }
    }
}

编辑：根据您提供的详细信息，您可以使用您自己的枚举器从外部库包装enumerator，如下所示，以便您可以在其上使用Parallel.ForEach：

我们假设你得到的枚举器类似于MockEnumerator，我们将它包装在一个普通的 IEnumerator 和 IEnumerable 中，这样我们就可以使用Parallel.ForEach 来并行读取。

class Program
{
    class Item
    {
        public int SomeProperty { get; }

        public Item(int prop)
        {
            SomeProperty = prop;
        }
    }

    class MockEnumerator
    {
        private Item[] _items = new Item[] { new Item(1), new Item(2) };
        private int _position = 0;

        public bool Next()
        {
            return _position++ < _items.Length;
        }

        public Item Read()
        {
            return _items[_position];
        }
    }

    class EnumeratorWrapper : IEnumerator<Item>, IEnumerable<Item>
    {
        private readonly MockEnumerator _enumerator;

        public EnumeratorWrapper(MockEnumerator enumerator)
        {
            this._enumerator = enumerator;
        }

        public Item Current => _enumerator.Read();

        object IEnumerator.Current => Current;

        public void Dispose()
        {
        }

        public IEnumerator<Item> GetEnumerator()
        {
            throw new NotImplementedException();
        }

        public bool MoveNext()
        {
            return _enumerator.Next();
        }

        public void Reset()
        {
        }

        IEnumerator IEnumerable.GetEnumerator()
        {
            return this;
        }
    }

    private static List<Item> _list = new List<Item>();

    static void Main(string[] args)
    {
        var enumerator = new EnumeratorWrapper(new MockEnumerator());
        Parallel.ForEach(enumerator, item =>
        {
            if (item.SomeProperty == 1)//someval
            {
                lock (_list)
                {
                    _list.Add(item);
                }
            }
        });
    }
}

【讨论】：

【解决方案2】：

这是基于任务的并行化的一个很好的例子。一个项目的每个处理对应一个任务。因此，您可以将循环更改为以下内容：

  var tasks = new List<Task<int>>();
  while(enumerator.MoveNext())
  {
    var item = enumerator.Current;

    Task<int> task = new Task<int>(() => ProcessItem(item));
    task.Start();
    tasks.Add(task);
  }


  foreach(Task<int> task in tasks)
  {
    int i = task.Result;
    classList.Add(i);
  }

请注意，classList 上的同步是通过首先在 while 循环中生成所有任务然后在 foreach 循环中合并结果来隐式给出的。同步是通过访问Result 和waits until the corresponding task is finished 来实现的。

【讨论】：