Замените Task.WhenAll на PLinq - PullRequest
Новая
       42

Замените Task.WhenAll на PLinq

0 голосов
/ 21 января 2020

У меня есть метод, который вызывает службу WCF несколько раз параллельно. Чтобы предотвратить перегрузку в целевой системе, я хочу использовать способность PLinq ограничивать количество параллельных выполнений. Теперь мне интересно, как я мог бы эффективно переписать свой метод.

Вот моя текущая реализация: 

private async Task RunFullImport(IProgress<float> progress) {
    var dataEntryCache = new ConcurrentHashSet<int>();
    using var client = new ESBClient(); // WCF

    // Progress counters helpers
    float totalSteps = 1f / companyGroup.Count();
    int currentStep = 0;

    //Iterate over all resources
    await Task.WhenAll(companyGroup.Select(async res => {
        getWorkOrderForResourceDataSetResponse worResp = await client.getWorkOrderForResourceDataSetAsync(
            new getWorkOrderForResourceDataSetRequest(
                "?",
                res.Company,
                res.ResourceNumber,
                res.ResourceType,
                res.CLSName,
                fromDate,
                toDate,
                "D"
            )
        );

        // Iterate over all work orders and add them to the list
        foreach (dsyWorkOrder02TtyWorkOrderResource workOrder in worResp.dsyWorkOrder02) {
            dataEntryCache.Add(workOrder.DataEntryNumber.Value);
        }

        // Update progress
        progress.Report(totalSteps * Interlocked.Increment(ref currentStep) * .1f);
    }));

    // Do some more stuff with the result
}

Я уже пробовал некоторые подходы, чтобы использовать PLinq вместо этого, но я не совсем придумать правильное решение. Это мое текущее состояние. 

var p = companyGroup.Select(res => client.getWorkOrderForResourceDataSetAsync(
    new getWorkOrderForResourceDataSetRequest(
        "?",
        res.Company,
        res.ResourceNumber,
        res.ResourceType,
        res.CLSName,
        fromDate,
        toDate,
        "D"
    )
).ContinueWith(worResp => {
    // Iterate over all work orders and add them to the list
    foreach (dsyWorkOrder02TtyWorkOrderResource workOrder in worResp.Result.dsyWorkOrder02) {
        dataEntryCache.Add(workOrder.DataEntryNumber.Value);
    }

    // Update progress
    progress.Report(totalSteps * Interlocked.Increment(ref currentStep) * .1f);
}));

await Task.WhenAll(p.AsParallel().ToArray());

Это, очевидно, не работает должным образом. Есть ли у вас какие-либо предложения, чтобы заставить его работать должным образом и эффективно и ограничить максимальное количество вызовов к серверу так 8 параллельно?

1 Ответ

3 голосов
/ 21 января 2020

PLINQ работает только с синхронным кодом. Он имеет несколько хороших встроенных регуляторов для управления количеством одновременных параллельных операций.

Чтобы контролировать количество одновременных асинхронных операций, используйте SemaphoreSlim: 

private async Task RunFullImport(IProgress<float> progress) {
  var dataEntryCache = new ConcurrentHashSet<int>();
  using var client = new ESBClient(); // WCF
  var limiter = new SemaphoreSlim(10); // or however many you want to limit to.

  // Progress counters helpers
  float totalSteps = 1f / companyGroup.Count();
  int currentStep = 0;

  //Iterate over all resources
  await Task.WhenAll(companyGroup.Select(async res => {
    await limiter.WaitAsync();
    try {    
      getWorkOrderForResourceDataSetResponse worResp = ...

      // Iterate over all work orders and add them to the list
      foreach (dsyWorkOrder02TtyWorkOrderResource workOrder in worResp.dsyWorkOrder02) {
        dataEntryCache.Add(workOrder.DataEntryNumber.Value);
      }

      // Update progress
      progress.Report(totalSteps * Interlocked.Increment(ref currentStep) * .1f);
    }
    finally {
      limiter.Release();
    }
  }));

  // Do some more stuff with the result
}

Для получения дополнительной информации см. Рецепт 12.5 в моей книге , который охватывает несколько различных решений для регулирования.

...