Reescritura utilizando foreach IObservable y Marco reactiva

Question

Estoy en VS2008 con Entity Framework. Estoy acceso a los objetos de la base de datos utilizando esql para DONDE EN funcionalidad. Estoy pasando una tonelada de ID para la instrucción de selección por lo que trozo arriba en grupos de 800. Entonces se funden juntos los resultados de cada trozo. Mi objetivo es la obtención de resultados para cada trozo en paralelo, en lugar de esperar de forma sincrónica. He instalado Marco reactiva y estoy bastante seguro de que tengo que hacer uso de ForkJoin. Sin embargo, no puedo encontrar la manera de convertir esta función para utilizarla. Aquí está mi código existente:

    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;
    }

Gracias!

  

    

      

EDITAR >>>       He modificado el código para utilizar de la siguiente manera 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;
    }

Se trabajó mucho la primera vez. Pero la segunda vez que me encontré con él, me dio este error:

La conexión no se ha cerrado. estado actual de la conexión está conectando. Descripción: Una excepción no controlada durante la ejecución de la solicitud Web actual. Revise el seguimiento de la pila para obtener más información acerca del error y dónde se originó en el código.

Detalles de la excepción: System.InvalidOperationException: La conexión no se ha cerrado. estado actual de la conexión está conectando.

Fuente de error:

Línea 49: foreach (var IAsyncResult en resultList) Línea 50: { Línea 51: del.EndInvoke (IAsyncResult); Línea 52: iAsyncResult.AsyncWaitHandle.Close (); Línea 53:}

Es interesante, b / c Emre (el autor de la biblioteca) tiene una edición a su puesto original, hablando de cómo agregó esas líneas de código para una mayor seguridad. ¿Estoy usando bien? O era su v1 más seguro después de todo?

Answer 1

VS2010 tiene que con PLINQ. Utilizando el AsParallel().WithDegreeOfParallelism(nbProcessors) extensiones haría lo que necesita.

Con VS2008, he utilizado del pobre hombre Parallel.ForEach iterador de Emre Aydinceren en el pasado cuando yo estaba tratando de evitar un cuello de botella, trate de darle un tiro.

EDIT: Como reacción al error que ha añadido, que podría ser un tiro al azar en la oscuridad, pero los contextos separados para cada hilo? De esta manera:

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());
    });

Es posible que tenga que ajustar alrededor de algunas cosas (como tomar la cadena Connextion del Contexto extendido para crear instancias de nuevos contextos).