使用 IObservable 和 Reactive Framework 重写 foreach

Question

我在 VS2008 中使用实体框架。我正在使用 esql 访问数据库中的对象以获得 WHERE IN 功能。我将大量 ID 传递给 select 语句，因此我将其分成 800 个组。然后我将每个块的结果合并在一起。我的目标是并行获取每个块的结果，而不是同步等待。我安装了 Reactive Framework，并且很确定我需要使用 ForkJoin。但是，我不知道如何转换这个函数来使用它。这是我现有的代码：

    public static IList<TElement> SelectWhereIn<TElement, TValue>(this ObjectContext context, string fieldName, IList<TValue> idList)
    {
        var chunkedIds = idList.Split(CHUNK_SIZE);
        string entitySetName = typeof(TElement).Name + "Set";
        var retList = new List<TElement>();
        foreach (var idChunk in chunkedIds)
        {
            string delimChunk = string.Join(",", idChunk.Select(x => x.ToString()).ToArray());
            ObjectQuery<TElement> query = context.CreateQuery<TElement>("SELECT VALUE x FROM " + entitySetName + " AS x");
            query = query.Where("it." + fieldName + " IN {" + delimChunk + "}");
            retList.AddRange(query);
        }
        return retList;
    }

谢谢！

编辑 >>> 我修改了代码以使用Poor Man，如下所示：

    public static IList<TElement> SelectWhereIn<TElement, TValue>(this ObjectContext context, string fieldName, IList<TValue> idList)
    {
        var chunkedIds = idList.Split(CHUNK_SIZE);
        string entitySetName = typeof(TElement).Name + "Set";
        var chunkLists = new List<IEnumerable<TElement>>();
        Parallel.ForEach(chunkedIds, idChunk =>
        {
            string delimChunk = string.Join(",", idChunk.Select(x => x.ToString()).ToArray());
            ObjectQuery<TElement> query = context.CreateQuery<TElement>("SELECT VALUE x FROM " + entitySetName + " AS x");
            query = query.Where("it." + fieldName + " IN {" + delimChunk + "}");
            chunkLists.Add(query.ToList());
        });
        var retList = new List<TElement>();
        foreach (var chunkList in chunkLists)
        {
            retList.AddRange(chunkList);
        }
        return retList;
    }

第一次效果很好。但是我第二次运行它时，我得到了这个错误：

连接没有关闭。连接的当前状态是正在连接。 说明：执行当前 Web 请求期间发生未处理的异常。请查看堆栈跟踪以获取有关错误及其源自代码的位置的更多信息。

异常详细信息：System.InvalidOperationException：连接未关闭。连接的当前状态为正在连接。

来源错误：

第 49 行：foreach（结果列表中的变量 iAsyncResult） 第 50 行：{ 第 51 行：del.EndInvoke(iAsyncResult); 第 52 行：iAsyncResult.AsyncWaitHandle.Close(); 第 53 行：}

有趣的是，b/c Emre（该库的作者）对他的原始帖子进行了编辑，讨论了他如何添加这些代码行以增加安全性。我用对了吗？还是他的 v1 更安全？

Answer 1

VS2010 确实具有 PLINQ 的功能。使用扩展 AsParallel().WithDegreeOfParallelism(nbProcessors) 可以满足您的需求。

对于 VS2008，我过去在尝试解决性能瓶颈时使用过 Poor Man's Parallel.ForEach Iterator by Emre Aydinceren，尝试尝试一下。

编辑：作为对您添加的错误的反应，它可能是在黑暗中随机拍摄，但每个线程都有单独的上下文？像这样：

Parallel.ForEach(chunkedIds, idChunk =>
    {
        ObjectContext context = new MyContext(connStr);//depending what's your config
                                                       // like, with or w/o conn string

        string delimChunk = string.Join(",", idChunk.Select(x => x.ToString()).ToArray());
        ObjectQuery<TElement> query = context.CreateQuery<TElement>("SELECT VALUE x FROM " + entitySetName + " AS x");
        query = query.Where("it." + fieldName + " IN {" + delimChunk + "}");
        chunkLists.Add(query.ToList());
    });

您可能需要调整一些事情（例如从扩展的上下文中获取连接字符串以实例化新上下文）。