Как я могу использовать "Where" с предикатом async?

У меня есть метод асинхронного предиката, например:

private async Task<bool> MeetsCriteria(Uri address)
{
    //Do something involving awaiting an HTTP request.
}

Скажем, у меня есть коллекция Uri s:

var addresses = new[]
{
    new Uri("http://www.google.com/"),
    new Uri("http://www.stackoverflow.com/") //etc.
};

Я хочу фильтровать addresses с помощью MeetsCriteria. Я хочу сделать это асинхронно; Я хочу, чтобы множественные вызовы предиката выполнялись асинхронно, и я хочу, чтобы они дождались завершения их и создания фильтрованного набора результатов. К сожалению, LINQ не поддерживает асинхронные предикаты, поэтому что-то вроде не работает:

var filteredAddresses = addresses.Where(MeetsCriteria);

Есть ли такой же удобный способ сделать это?

Ответы

Ответ 1

Я думаю, что одна из причин, почему это не так, заключается в том, что существует множество возможных вариантов, и каждый выбор будет правильным при определенных обстоятельствах:

  • Если предикаты выполняются параллельно или последовательно?
    • Если они выполняются параллельно, должны ли они выполняться сразу или должна быть ограничена степень parallelism?
    • Если они выполняются параллельно, должны ли результаты быть в том же порядке, что и исходная коллекция, в порядке завершения или в undefined порядке?
      • Если они должны быть возвращены в порядке завершения, должен ли быть какой-то способ (асинхронно) получить результаты по мере их завершения? (Для этого потребуется изменение типа возврата из Task<IEnumerable<T>> в другое.)

Вы сказали, что хотите, чтобы предикаты выполнялись параллельно. В этом случае самым простым выбором является выполнить их все сразу и вернуть их в порядке завершения:

static async Task<IEnumerable<T>> Where<T>(
    this IEnumerable<T> source, Func<T, Task<bool>> predicate)
{
    var results = new ConcurrentQueue<T>();
    var tasks = source.Select(
        async x =>
        {
            if (await predicate(x))
                results.Enqueue(x);
        });
    await Task.WhenAll(tasks);
    return results;
}

Затем вы можете использовать его следующим образом:

var filteredAddresses = await addresses.Where(MeetsCriteria);

Ответ 2

Первый подход: выдать все запросы вперед один за другим, затем ждать, пока все запросы вернутся, а затем отфильтруйте результат. (svick code также сделал это, но здесь я делаю это без промежуточного ConcurrentQueue).

// First approach: massive fan-out
var tasks = addresses.Select(async a => new { A = a, C = await MeetsCriteriaAsync(a) });
var addressesAndCriteria = await Task.WhenAll(tasks);
var filteredAddresses = addressAndCriteria.Where(ac => ac.C).Select(ac => ac.A);

Второй подход: делать запросы один за другим. Это займет больше времени, но оно не будет забивать веб-сервис с огромным натиском запросов (предполагая, что MeetsCriteriaAsync выходит на веб-сервис...)

// Second approach: one by one
var filteredAddresses = new List<Uri>();
foreach (var a in filteredAddresses)
{
  if (await MeetsCriteriaAsync(a)) filteredAddresses.Add(a);
}

Третий подход: как и для второго, но с использованием гипотетической функции С# 8 "асинхронные потоки". С# 8 еще не вышел, а асинхронные потоки еще не разработаны, но мы можем мечтать! Тип IAsyncEnumerable уже существует в RX, и, надеюсь, они добавят для него больше комбинаторов. Хорошая вещь об IAsyncEnumerable заключается в том, что мы можем начать употреблять первые несколько фильтрованныхAddresses, как только они придут, а не ждать, пока все будет отфильтровано в первую очередь.

// Third approach: ???
IEnumerable<Uri> addresses = {...};
IAsyncEnumerable<Uri> filteredAddresses = addresses.WhereAsync(MeetsCriteriaAsync);

Четвертый подход: возможно, мы не хотим забивать webservice всеми запросами сразу, но мы рады выпустить более одного запроса за раз. Возможно, мы провели эксперименты и обнаружили, что "три за раз" были счастливой средой. ПРИМЕЧАНИЕ. Этот код предполагает однопоточный контекст выполнения, такой как программирование пользовательского интерфейса или ASP.NET. Если он запускается в многопоточном контексте выполнения, ему требуется ConcurrentQueue и ConcurrentList.

// Fourth approach: throttle to three-at-a-time requests
var addresses = new Queue<Uri>(...);
var filteredAddresses = new List<Uri>();
var worker1 = FilterAsync(addresses, filteredAddresses);
var worker2 = FilterAsync(addresses, filteredAddresses);
var worker3 = FilterAsync(addresses, filteredAddresses);
await Task.WhenAll(worker1, worker2, worker3);

async Task FilterAsync(Queue<Uri> q, List<Uri> r)
{
  while (q.Count > 0)
  {
    var item = q.Dequeue();
    if (await MeetsCriteriaAsync(item)) r.Add(item);
  }
}

Есть способы сделать четвертый подход, используя библиотеку потока данных TPL.