c# - 在IEnumerable实现中使用任务并行库来提高速度

以下代码是我尝试优化的代码的简化版本。

void Main()
{
    var words = new List<string> {"abcd", "wxyz", "1234"};

    foreach (var character in SplitItOut(words))
    {
        Console.WriteLine (character);
    }
}

public IEnumerable<char> SplitItOut(IEnumerable<string> words)
{
    foreach (string word in words)
    {
        var characters = GetCharacters(word);

        foreach (char c in characters)
        {
            yield return c;
        }
    }
}

char[] GetCharacters(string word)
{
    Thread.Sleep(5000);
    return word.ToCharArray();
}

我无法更改方法SplitItOut的签名.GetCharacters方法的调用成本很高，但是是线程安全的。 SplitItOut方法的输入可以包含100,000多个条目，并且对GetCharacters()方法的单次调用可能需要200毫秒左右。它还会引发我可以忽略的异常。结果的顺序无关紧要。

在我的第一次尝试中，我想出了以下使用TPL的实现，该实现可以大大加快速度，但是一直阻塞直到我完成所有单词的处理。

public IEnumerable<char> SplitItOut(IEnumerable<string> words)
{
    Task<char[][]> tasks = Task<char[][]>.Factory.StartNew(() =>
    {
        ConcurrentBag<char[]> taskResults = new ConcurrentBag<char[]>();

        Parallel.ForEach(words,
            word =>
            {
                taskResults.Add(GetCharacters(word));
            });

        return taskResults.ToArray();
    });

    foreach (var wordResult in tasks.Result)
    {
        foreach (var c in wordResult)
        {
            yield return c;
        }
    }
}

我正在寻找比这更好的方法SplitItOut()的实现。缩短处理时间是我的首要任务。

最佳答案

如果我正确地阅读了您的问题，那么您不是要加快从单词创建字符的并行处理的速度-您希望您的枚举数能够在准备好每个单词后立即产生。使用您当前拥有的实现(以及我目前看到的其他答案)，SplitItOut将等待，直到将所有单词发送到GetCharacters，并在生成第一个单词之前返回所有结果。

在这种情况下，我喜欢将事情分解为生产者和消费者的过程。生产者线程将使用可用的单词并调用GetCharacters，然后将结果转储到某个地方。使用者一旦准备好，就会向SplitItOut的调用者提供字符。确实，消费者是SplitItOut的调用方。

我们可以将 BlockingCollection 既用作产生字符的方式，又用作放置结果的“某处”。我们可以使用 ConcurrentBag 作为放置尚未拆分的单词的位置:

static void Main()
        {
            var words = new List<string> { "abcd", "wxyz", "1234"};

            foreach (var character in SplitItOut(words))
            {
                Console.WriteLine(character);
            }
        }


        static char[] GetCharacters(string word)
        {
            Thread.Sleep(5000);
            return word.ToCharArray();
        }

无需更改main或GetCharacters-因为这些表示您的约束(无法更改调用方，无法更改昂贵的操作)

        public static IEnumerable<char> SplitItOut(IEnumerable<string> words)
        {
            var source = new ConcurrentBag<string>(words);
            var chars = new BlockingCollection<char>();

            var tasks = new[]
                   {
                       Task.Factory.StartNew(() => CharProducer(source, chars)),
                       Task.Factory.StartNew(() => CharProducer(source, chars)),
                       //add more, tweak away, or use a factory to create tasks.
                       //measure before you simply add more!
                   };

            Task.Factory.ContinueWhenAll(tasks, t => chars.CompleteAdding());

            return chars.GetConsumingEnumerable();
        }

在这里，我们将SplitItOut方法更改为要做四件事:

用我们希望拆分的所有单词初始化并发包。 (旁注:如果您想按需枚举单词，可以启动一个新任务将其插入，而不是在构造函数中执行该操作)

启动我们的char“生产者”任务。您可以开始设置号码，使用工厂等。我建议您在测量之前不要疯狂。

在完成所有任务后，用信号通知我们我们完成了BlockingCollection。

“消耗”所有产生的字符(我们可以轻松完成自己的工作，只返回一个IEnumerable<char>而不是foreach和yield，但是如果您愿意的话，您可以做很多事情)

缺少的只是我们的生产者实现。我已经扩展了所有linq快捷方式以使其清晰可见，但是它非常简单:

        private static void CharProducer(ConcurrentBag<string> words, BlockingCollection<char> output)
        {
            while(!words.IsEmpty)
            {
                string word;
                if(words.TryTake(out word))
                {
                    foreach (var c in GetCharacters(word))
                    {
                        output.Add(c);
                    }
                }
            }
        }

这只是

从ConcurrentBag中取出一个单词(除非它是空的-如果是，则说明任务已完成!)

调用昂贵的方法

将输出放入BlockingCollection