IObservable 및 Reactive Framework를 사용하여 foreach 다시 작성

Question

저는 Entity Framework를 사용하는 VS2008을 사용하고 있습니다.WHERE IN 기능을 위해 esql을 사용하여 데이터베이스의 개체에 액세스하고 있습니다.Select 문에 수많은 ID를 전달하므로 이를 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;
    }

감사해요!

편집 >>> 아래와 같이 가난한 사람을 사용하도록 코드를 수정했습니다.

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

처음에는 훌륭하게 작동했습니다.그런데 두 번째로 실행했을 때 다음과 같은 오류가 발생했습니다.

연결이 닫히지 않았습니다.연결의 현재 상태는 연결 중입니다.설명:현재 웹 요청을 실행하는 동안 처리되지 않은 예외가 발생했습니다.오류와 오류가 발생한 코드 위치에 대한 자세한 내용은 스택 추적을 검토하세요.

예외 세부정보:System.InvalidOperation예외:연결이 닫히지 않았습니다.연결의 현재 상태는 연결 중입니다.

소스 오류:

49행:foreach (var iasyncresult in resultlist) 줄 50 :{라인 51 :del.EndInvoke(iAsyncResult);52행:iAsyncResult.AsyncWaitHandle.Close();53행:}

흥미롭습니다. b/c Emre(라이브러리 작성자)는 안전을 강화하기 위해 해당 코드 줄을 추가한 방법에 대해 설명하는 원래 게시물을 편집했습니다.내가 제대로 사용하고 있는 걸까?아니면 결국 그의 v1이 더 안전했습니까?

Answer 1

VS2010에는 PLINQ가 있습니다.확장 프로그램 사용 AsParallel().WithDegreeOfParallelism(nbProcessors) 당신이 필요한 것을 할 것입니다.

VS2008에서 나는 가난한 사람의 Parallel.ForEach 반복자(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());
    });

몇 가지 사항을 조정해야 할 수도 있습니다(새 컨텍스트를 인스턴스화하기 위해 확장된 컨텍스트에서 연결 문자열을 가져오는 것과 같은).