데이터가 적을 때만 비동기

Question

푸른 테이블의 데이터 - 약 5k 개의 테이블을 읽고 다른 메트릭을 수집하고이를 다른 비어있는 테이블에 저장합니다. 비동기 방식으로 모든 것을 저장합니다. 내가 직면하고있는 문제는 때로는 일어날 수있는 거대한 데이터가있을 때 응용 프로그램이 멈추는 것입니다. 동일한 코드가 적은 데이터로도 잘 작동합니다. 내가 뭐하는 거지 단계는 (그들 모두 수신이 비동기 사용하여 비동기 및 기다리고 있습니다)

1. 1 & 2 병렬에있는 모든 테이블을 이전 메트릭 데이터를 (읽기 푸른
2. 에서 모든 테이블 이름을 읽기 - Task.WhenAll)
3. 각 테이블 프로세스에서 데이터를 가져 오기 및 저장 (Task.WhenAll)
무엇인지 내가 원하는

, 그것까지 사용 asynchronousy 내 응용 프로그램 매달려하지 않습니다. 처리 할 수있는 것보다 많은 데이터가있는 경우, 더 이상 사용할 수있는 데이터 처리를 완료하는 데 더 이상 테이블의 데이터를 읽지 않아야합니다.

Parallel.ForEach이 처리합니까?

코드 : 은 Stephen Cleary으로 편집되었습니다. 아직 모든 테이블에서 작동하지 않습니다. 반면 그것은 500 테이블을 위해 일하고있다.

나는 스레드의 수보다는 정지 상태로 애플 리케이션 (콘솔 어플리케이션)을 가져 오는 데이터의 양이라고 생각한다. (하나의 쓰레드는 백만 개의 행을 검색 할 수 있으며, 수 천 개가 넘고 메서드가 넘겨지면 그 수는 사전에 추가되므로 더 많은 메모리가 필요하면 가비지 수집이 가능하다.) 또는 구현 한 방법이다 Semaphoreslim 그게 잘못된거야? 당신의 async 이후

public async Task CalculateMetricsForAllTablesAsync() 
{ 
    var allWizardTableNamesTask = GetAllWizardTableNamesAsync(); 
    var allTablesNamesWithLastRunTimeTask = GetAllTableNamesWithLastRunTimeAsync(); 

    await Task.WhenAll(allWizardTableNamesTask, allTablesNamesWithLastRunTimeTask).ConfigureAwait(false); 

    var allWizardTableNames = allWizardTableNamesTask.Result; 
    var allTablesNamesWithLastRunTime = allTablesNamesWithLastRunTimeTask.Result; 

    var throttler = new SemaphoreSlim(10); 
    var concurrentTableProcessingTasks = new ConcurrentStack<Task>(); 

    foreach (var tname in allWizardTableNames) 
    { 
     await throttler.WaitAsync(); 
     try 
     { 
      concurrentTableProcessingTasks.Push(ProcessTableDataAsync(tname, getTableNameWithLastRunTime(tname))); 
     } 
     finally 
     { 
      throttler.Release(); 
     } 
    } 

    await Task.WhenAll(concurrentTableProcessingTasks).ConfigureAwait(false); 

} 

private async Task ProcessTableDataAsync(string tableName, Tuple<string, string> matchingTable) 
{ 
    var tableDataRetrieved = new TaskCompletionSource<bool>(); 
    var metricCountsForEachDay = new ConcurrentDictionary<string, Tuple<int, int>>(); 

    _fromATS.GetTableDataAsync<DynamicTableEntity>(tableName, GetFilter(matchingTable)) 
     .Subscribe(entities => ProcessWizardDataChunk(metricCountsForEachDay, entities),() => tableDataRetrieved.TrySetResult(true)); 

    await tableDataRetrieved.Task; 
    await SaveMetricDataAsync(tableName, metricCountsForEachDay).ConfigureAwait(false); 
} 

Answer 1

내가 async 수준에서 조절 권하고 싶습니다, 수신을 포장한다. SemaphoreSlim을 정의하고 메소드 논리를 WaitAsync/Release 내에 랩핑하여이를 수행 할 수 있습니다.

또는 TPL 데이터 흐름을 고려하십시오. Dataflow에는 스로틀 링을위한 옵션 (MaxDegreeOfParallelism)이 내장되어 있으며 async 및 Rx와 자연스럽게 상호 운용됩니다.