List<T>.RemoveAll并行

Question

我想知道做toProcess.RemoveAll的另一种选择，但是并行的。今天，我的代码就像我的例子一样工作得很好，但是在顺序上，我想要做得很好。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace ParallelTest
{
    using System.Threading;
    using System.Threading.Tasks;
    class Program
    {
        static void Main(string[] args)
        {
            List<VerifySomethingFromInternet> foo = new List<VerifySomethingFromInternet>();
            foo.Add(new VerifySomethingFromInternet(@"id1", true));
            foo.Add(new VerifySomethingFromInternet(@"id2", false));
            foo.Add(new VerifySomethingFromInternet(@"id3", true));
            foo.Add(new VerifySomethingFromInternet(@"id4", false));
            foo.Add(new VerifySomethingFromInternet(@"id5", true));
            foo.Add(new VerifySomethingFromInternet(@"id6", false));

            DoSomethingFromIntert bar = new DoSomethingFromIntert();

            bar.DoesWork(foo);
            Console.ReadLine();
        }
    }

    public class DoSomethingFromIntert
    {
        bool RemoveIFTrueFromInternet(VerifySomethingFromInternet vsfi)
        {
            Console.WriteLine(String.Format("Identification : {0} - Thread : {1}", vsfi.Identification, Thread.CurrentThread.ManagedThreadId));
            // Do some blocking work at internet
            return vsfi.IsRemovable;
        }

        public void DoesWork(List<VerifySomethingFromInternet> toProcess)
        {
            Console.WriteLine(String.Format("total : {0}", toProcess.Count));
            //Remove all true return
            toProcess.RemoveAll(f => this.RemoveIFTrueFromInternet(f));
            Console.WriteLine(String.Format("total : {0}", toProcess.Count));
        }
    }
    public class VerifySomethingFromInternet
    {
        public VerifySomethingFromInternet(string id, bool remove)
        {
            this.Identification = id;
            this.IsRemovable = remove;
        }
        public string Identification { get; set; }
        public bool IsRemovable { get; set; }
    }
}

Answer 1

var newList = toProcess.AsParallel ()
               .Where (f => !this.RemoveIFTrueFromInternet(f))
               .ToList ();

toProcess = newList;

也许这回答了你的问题，但我不确定它是否真的更快。试着量一下。

请注意，这可能会更改列表中元素的顺序。如果您关心订单，请在AsOrdered后面添加AsParallel。(感谢威斯顿的暗示)。

Answer 2

List<T>并不是线程安全的，因此无法与这种类型的列表并行执行此操作。

您可以使用线程安全的ConcurrentBag代替，但是这个方法显然没有RemoveAll方法。

还可以将列表转换为数组，编辑该数组，然后再将其传递给list。

Answer 3

我试图对您的代码进行一些重构

我使用BlockingCollection实现生产者消费者方案。

这不是并行删除，但它可以通过并行处理来解决您的问题，尝试一下您可能会喜欢它。

class Program
{
    static void Main(string[] args)
    {
        DoSomethingFromIntert bar = new DoSomethingFromIntert();  
        bar.Verify(@"id1", true);
        bar.Verify(@"id2", false);
        bar.Verify(@"id3", true);
        bar.Verify(@"id4", false);
        bar.Verify(@"id5", true);
        bar.Verify(@"id6", false);

        bar.Complete();

        Console.ReadLine();
    }
}

public class DoSomethingFromIntert
{
    BlockingCollection<VerifySomethingFromInternet> toProcess = new BlockingCollection<VerifySomethingFromInternet>();
    ConcurrentBag<VerifySomethingFromInternet> workinglist = new ConcurrentBag<VerifySomethingFromInternet>();

    public DoSomethingFromIntert()
    {
        //init four consumers you may choose as many as you want
        ThreadPool.QueueUserWorkItem(DoesWork);
        ThreadPool.QueueUserWorkItem(DoesWork);
        ThreadPool.QueueUserWorkItem(DoesWork);
        ThreadPool.QueueUserWorkItem(DoesWork);
    }

    public void Verify(string param, bool flag)
    {
        //add to the processing list
        toProcess.TryAdd(new VerifySomethingFromInternet(param, flag));
    }

    public void Complete()
    {
        //mark producer as complete and let the threads exit when finished verifying
        toProcess.CompleteAdding();
    }

    bool RemoveIFTrueFromInternet(VerifySomethingFromInternet vsfi)
    {
        Console.WriteLine(String.Format("Identification : {0} - Thread : {1}", vsfi.Identification, Thread.CurrentThread.ManagedThreadId));
        // Do some blocking work at internet
        return vsfi.IsRemovable;
    }

    private void DoesWork(object state)
    {
        Console.WriteLine(String.Format("total : {0}", toProcess.Count));
        foreach (var item in toProcess.GetConsumingEnumerable())
        {
            //do work
            if (!RemoveIFTrueFromInternet(item))
            {
                //add to list if working
                workinglist.TryAdd(item);
            }
            //no need to remove as it is removed from the list automatically
        }
        //this line will only reach after toProcess.CompleteAdding() and when items are consumed(verified)
        Console.WriteLine(String.Format("total : {0}", toProcess.Count));
    }
}

简而言之，一旦您添加了这些项，它就会开始验证它们，并将成功的项保存在单独的列表中。

编辑

由于GetConsumingEnumerable()的foreach循环在默认情况下不会结束，它一直在等待下一个元素，直到调用CompleteAdding()为止。因此，我在包装器类中添加了Complete()方法，以在我们推送所有元素之后完成验证循环。

这样做的目的是继续向类中添加验证元素，并让使用者循环并行地验证每个元素，一旦完成，所有元素都将调用Complete()来知道没有更多的元素要添加，这样一旦列表为空，他们就可以终止foreach循环。

在您的代码中，删除元素不是性能的实际问题，而是验证过程的同步循环(如果是热点)。从列表中删除只需几毫秒的成本，但是代码中昂贵的部分是blocking work at internet，所以如果我们能够使它并行，我们就可以减少一些宝贵的时间。

请注意您初始化的使用者线程的数量，但是我使用了线程池，但是如果过度使用，仍然可能会影响性能。因此，根据机器的能力来决定一个数字。数目或核心/处理器

更多关于BlockingCollection的信息