Как выполнить несколько асинхронных вызовов параллельно?

Question

У меня есть ряд команд, которые делают вызовы к мыльному веб-сервису (API Betfair).Все они относятся к классической модели модели асинхронного программирования ...

public void DoXXX( <input parameters ...> )
{
    XXXRequest Request = new XXXRequest();
    // populate Request from input parameters ...
    BetfairService.BeginXXX( Request, XXXCallback, State );
}

private void XXXCallback(IAsyncResult Result)
{
    XXXResponse Response = BetfairService.EndXXX(Result);
    if (Response.ErrorCode == XXXErrorCode.OK)
        // store data from Response
    else
        // deal with error
}

Я хочу выполнить определенный набор команд, а затем выполнить некоторые вычисления, используя объединенные значения возвращаемых данных, после завершения всех команд.

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

Мое идеальное решение состояло бы в том, чтобы все эти команды выполнялись параллельно, а затем запускать вычисления после завершения всех команд.Я пытался просмотреть Task.Factory.FromAsync (), но все примеры, которые я могу найти, включают только прямые вызовы BeginXXX / EndXXX, ничего не делая с ответом.

У кого-нибудь есть указатели дляподходящее решение этой проблемы?

Answer 1

Чтобы использовать FromAsync, необходимо указать тип возвращаемого значения:

var task = Task<XXXResponse>.Factory.FromAsync( ...

Затем у вас есть задача со свойством Result типа XXXResponse.

Youзатем можно использовать Parallel.Invoke для параллельного запуска начальных команд.Это будет блокировать, пока все эти задачи не будут выполнены.Затем вы можете выполнить «дополнительную обработку».

Или же вы можете сохранить начальные задачи в массиве и использовать Task.Factory.ContinueWhenAll для создания продолжения.

Nick

Answer 2

Я бы посоветовал вам посмотреть на Microsoft Reactive Extensions (Rx), чтобы делать то, что вы хотите. Это позволяет вам превращать асинхронные операции (среди прочего) в наблюдаемые запросы LINQ.

Скажем, у меня есть три функции, каждая из которых требует значительного времени для вычисления:

Func<int> fa = () =>
{
    Thread.Sleep(2000);
    return 42;
};

Func<int, string, string> fb = (n, t) =>
{
    Thread.Sleep(n * 1000);
    return t + n.ToString();
};

Func<DateTimeOffset> fc = () =>
{
    Thread.Sleep(1000);
    return DateTimeOffset.UtcNow;
};

Затем я могу использовать метод FromAsyncPattern, чтобы превратить эти лямбда-функции в наблюдаемые функции:

Func<IObservable<int>> ofa =
    Observable
        .FromAsyncPattern<int>(
            fa.BeginInvoke,
            fa.EndInvoke);

Func<int, string, IObservable<string>> ofb =
    Observable
        .FromAsyncPattern<int, string, string>(
            fb.BeginInvoke,
            fb.EndInvoke);

Func<IObservable<DateTimeOffset>> ofc =
    Observable
        .FromAsyncPattern<DateTimeOffset>(
            fc.BeginInvoke,
            fc.EndInvoke);

Теперь я могу начать все вызовы, просто выполнив следующее:

IObservable<int> oa = ofa();
IObservable<string> ob = ofb(1, "foo");
IObservable<DateTimeOffset> oc = ofc();

Это эффективно запускает три вычисления параллельно. Теперь нам нужно просто свести результаты.

Здесь LINQ входит:

var query =
    from a in oa
    from b in ob
    from c in oc
    select new { a, b, c };

И тогда я подписываюсь на этот запрос, чтобы получить результаты:

query.Subscribe(p =>
{
    Console.WriteLine(p.a);
    Console.WriteLine(p.b);
    Console.WriteLine(p.c);
});

В своем тестировании я поместил таймеры вокруг этого кода для вычисления фактического времени выполнения. Даже если общее время должно составлять 4 секунды, если он выполняется последовательно, этот код заканчивается через 2 - максимальное время любого из трех.

Теперь этот пример - лишь малая часть того, что может сделать Rx, но это хорошая отправная точка.

Выкрикни, если я смогу что-нибудь объяснить дальше.

Вот ссылки для Rx:

• Форум реактивных расширений
• Реактивные расширения (Rx) v1.0.10621

Answer 3

У вас должен быть счетчик выполненных сервисных вызовов.В каждом методе обратного вызова вы должны проверять этот счетчик - если он равен максимальному количеству вызовов службы, вы должны выполнить дополнительную обработку, в противном случае - вы просто увеличиваете счетчик.