Riscrittura foreach utilizzando IObservable e quadro reattiva

Question

Sono in VS2008 con Entity Framework. Sto accedere agli oggetti dal database utilizzando ESQL per DOVE IN funzionalità. Sto passando un sacco di ID per l'istruzione select così ho pezzo in su in gruppi di 800. Poi ho fondere insieme i risultati di ogni pezzo. Il mio obiettivo è quello di ottenere i risultati per ogni pezzo in parallelo, piuttosto che aspettare in modo sincrono. Ho installato Reactive Framework e sono abbastanza sicuro che ho bisogno di fare uso di ForkJoin. Tuttavia, non riesco a capire come convertire questa funzione per usarlo. Ecco il mio codice esistente:

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

Grazie!

  

    

      

EDIT >>>       Ho modificato il codice per utilizzare povero di come di seguito:

    

  

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

Ha funzionato grande la prima volta. Ma la seconda volta che mi sono imbattuto, ho ottenuto questo errore:

La connessione non è stata chiusa. attuale stato della connessione è la connessione. Descrizione: Un'eccezione non gestita durante l'esecuzione della richiesta Web corrente. Si prega di rivedere l'analisi dello stack per ulteriori informazioni sull'errore e dove ha avuto origine nel codice.

Dettagli eccezione: System.InvalidOperationException: La connessione non è stata chiusa. attuale stato della connessione è la connessione.

Errore Origine:

Linea 49: foreach (var IAsyncResult in risultante a) Linea 50: { Linea 51: del.EndInvoke (IAsyncResult); Linea 52: iAsyncResult.AsyncWaitHandle.Close (); Linea 53:}

E 'interessante, b / c Emre (l'autore della biblioteca) ha una modifica al suo post originale parlando di come ha aggiunto quelle righe di codice per una maggiore sicurezza. Sto utilizzando nel modo giusto? O era il suo v1 più sicuro, dopo tutto?

Answer 1

VS2010 ha che con PLINQ. Utilizzando il AsParallel().WithDegreeOfParallelism(nbProcessors) estensioni avrebbe fatto quello che ti serve.

Con VS2008, ho usato Poverello di Parallel.ForEach Iterator da Emre Aydinceren in passato, quando stavo cercando di aggirare un collo di bottiglia, cercare di dare un colpo.

EDIT: Come reazione all'errore si è aggiunto, potrebbe essere un colpo casuale nel buio, ma i contesti separati per ogni thread? In questo modo:

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

Potrebbe essere necessario modificare intorno ad alcune cose (come prendere la stringa Connextion dal contesto estesa a un'istanza di nuovi contesti).