使用并行 Linq 扩展来合并两个序列，如何才能首先产生最快的结果？

假设我有两个返回整数 1 到 5 的序列。

第一个返回 1、2 和 3 的速度非常快，但 4 和 5 分别需要 200 毫秒。

public static IEnumerable<int> FastFirst()
{
    for (int i = 1; i < 6; i++)
    {
        if (i > 3) Thread.Sleep(200);
        yield return i;
    }
}

第二个返回 1、2 和 3，有 200 毫秒的延迟，但 4 和 5 的返回速度很快。

public static IEnumerable<int> SlowFirst()
{
    for (int i = 1; i < 6; i++)
    {
        if (i < 4) Thread.Sleep(200);
        yield return i;
    }
}

将这两个序列联合起来只得到数字 1 到 5。

FastFirst().Union(SlowFirst());

我无法保证这两种方法中的哪一种在哪一点出现延迟，因此执行顺序不能保证为我提供解决方案。因此，我想并行化联合，以最大限度地减少示例中的（人为）延迟。

真实场景：我有一个返回的缓存some实体和返回的数据源all实体。我希望能够从内部并行化对缓存和数据源的请求的方法返回迭代器，以便尽可能快地产生缓存的结果。

注 1：我意识到这仍然在浪费 CPU 周期；我不是问如何防止序列迭代其缓慢的元素，而是问如何尽快合并它们。

更新1：我定制了achitaka-san 的出色响应来接受多个生产者，并使用ContinueWhenAll 来设置BlockingCollection 的CompleteAdding 一次。我只是把它放在这里，因为它会因缺乏注释格式而丢失。任何进一步的反馈都会很棒！

public static IEnumerable<TResult> SelectAsync<TResult>(
    params IEnumerable<TResult>[] producer)
{
    var resultsQueue = new BlockingCollection<TResult>();

    var taskList = new HashSet<Task>();
    foreach (var result in producer)
    {
        taskList.Add(
            Task.Factory.StartNew(
                () =>
                    {
                        foreach (var product in result)
                        {
                            resultsQueue.Add(product);
                        }
                    }));
    }

    Task.Factory.ContinueWhenAll(taskList.ToArray(), x => resultsQueue.CompleteAdding());

    return resultsQueue.GetConsumingEnumerable();
}

看看这个。 第一个方法只是按照结果的顺序返回所有内容。 第二个检查唯一性。如果你把它们链接起来，我想你会得到你想要的结果。

public static class Class1
{
    public static IEnumerable<TResult> SelectAsync<TResult>(
        IEnumerable<TResult> producer1,
        IEnumerable<TResult> producer2,
        int capacity)
    {
        var resultsQueue = new BlockingCollection<TResult>(capacity);
        var producer1Done = false;
        var producer2Done = false;

        Task.Factory.StartNew(() =>
        {
            foreach (var product in producer1)
            {
                resultsQueue.Add(product);
            }
            producer1Done = true;
            if (producer1Done && producer2Done) { resultsQueue.CompleteAdding(); }
        });

        Task.Factory.StartNew(() =>
        {
            foreach (var product in producer2)
            {
                resultsQueue.Add(product);
            }
            producer2Done = true;
            if (producer1Done && producer2Done) { resultsQueue.CompleteAdding(); }
        });

        return resultsQueue.GetConsumingEnumerable();
    }


    public static IEnumerable<TResult> SelectAsyncUnique<TResult>(this IEnumerable<TResult> source)
    {
        HashSet<TResult> knownResults = new HashSet<TResult>();
        foreach (TResult result in source)
        {
            if (knownResults.Contains(result)) {continue;}
            knownResults.Add(result);
            yield return result;
        }
    }
}