Как сделать параллельные вызовы API в C #?

Question

Мне нужно получить пользовательские поля из API ипотеки.Проблема в том, что в общей сложности 11000 записей и это занимает 1 секунду на запрос API.Я хочу найти способ посылать запросы асинхронно и параллельно, чтобы сделать это более эффективным.

Я попытался выполнить цикл по всем запросам, тогда как Task.WaitAll() ожидал ответа для возврата.Я получаю только два ответа, после чего приложение ожидает бесконечно.

Сначала я установил статический класс для HttpClient

 public static class ApiHelper
    {
        public static HttpClient ApiClient { get; set; }

        public static void InitializeClient()
        {
            ApiClient = new HttpClient();
            ApiClient.DefaultRequestHeaders.Add("ContentType", "application/json");
        }
    }

. Я собираю свой список ипотечных идентификаторов и перебираю сообщение API.Звонки

        static public DataTable GetCustomFields(DataTable dt, List<string> cf, string auth)
        {

                //set auth header
                ApiHelper.ApiClient.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", auth);

                //format body
                string jsonBody = JArray.FromObject(cf).ToString();
                var content = new StringContent(jsonBody, Encoding.UTF8, "application/json");



                var responses = new List<Task<string>>();


                foreach (DataRow dr in dt.Rows)
                {

                    string guid = dr["GUID"].ToString().Replace("{", "").Replace("}", ""); //remove {} from string

                    responses.Add(GetData(guid, content));

                }

                Task.WaitAll(responses.ToArray());
                //some code here to process through the responses and return a datatable

                return updatedDT;

        }

Для каждого вызова API требуется идентификатор залога (GUID) в URL-адресе

  async static Task<string> GetData(string guid, StringContent json)
        {

            string url = "https://api.elliemae.com/encompass/v1/loans/" + guid + "/fieldReader";
            Console.WriteLine("{0} has started .....", guid);
            using (HttpResponseMessage response = await ApiHelper.ApiClient.PostAsync(url, json))
            {
                if (response.IsSuccessStatusCode)
                {
                    Console.WriteLine("{0} has returned response....", guid);
                    return await response.Content.ReadAsStringAsync();
                }
                else
                {
                    Console.WriteLine(response.ReasonPhrase);
                    throw new Exception(response.ReasonPhrase);
                }

            }

        }

Сейчас я проверяю только 10 записей и отправляю все 10 запросов.Но я получаю только два обратно.

Результат здесь .

Не могли бы вы посоветовать мне правильный способ отправки одновременных вызовов API?

Answer 1

Все GetData Task используют один и тот же экземпляр HttpClient singleton.HttpClient не может обслуживать несколько вызовов одновременно.Рекомендуется использовать Pool HttpClient, чтобы гарантировать, что нет Задачи, обращающейся к одному и тому же HttpClient одновременно.

Кроме того, будьте осторожны, если вы добавите exception в Task, это остановит WaitAll() при первом выдаваемом исключении

Решение Я разместил весь проект здесь:https://github.com/jonathanlarouche/stackoverflow_58137212
Это решение отправляет 25 запросов, используя пул max sized [3];

В основном ApiHelper содержит пул HttpClient , используя общий класс ArrayPool<T>. Вы можете использовать любую другую библиотеку пулинга, я просто хотел опубликовать автономное решение .

Рекомендуемый ApiHelper Ниже, теперь этот класссодержит пул и метод Use, которые получают Action, Предмет из пула будет «арендован» на время действия, а затем будет возвращен в пул через функцию ArrayPool.Use.Функция Use также получает apiToken для изменения заголовка авторизации запроса.

public static class ApiHelper
{
    public static int PoolSize { get => apiClientPool.Size; }

    private static ArrayPool<HttpClient> apiClientPool = new ArrayPool<HttpClient>(() => {
        var apiClient = new HttpClient();
        apiClient.DefaultRequestHeaders.Add("ContentType", "application/json");
        return apiClient;
    });

    public static Task Use(string apiToken, Func<HttpClient, Task> action)
    {
        return apiClientPool.Use(client => {
            client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", apiToken);
            return action(client);
        });
    }
}

GetData .Get Data получит apiToken и будет ожидать функцию ApiHelper.Use.В этой функции необходимо создать новый экземпляр объекта StringContent(), так как он не может быть повторно использован в различных вызовах Http Post.

async static Task<string> GetData(string apiToken, Guid guid, string jsonBody)
{

    string url = "https://api.elliemae.com/encompass/v1/loans/" + guid + "/fieldReader";
    Console.WriteLine("{0} has started .....", guid);
    string output = null;
    await ApiHelper.Use(apiToken, (client) => 
    {
        var json = new StringContent(jsonBody, Encoding.UTF8, "application/json");
        return client.PostAsync(url, json).ContinueWith(postTaskResult =>
        {

            return postTaskResult.Result.Content.ReadAsStringAsync().ContinueWith(s => {

                output = s.Result;
                return s;
            });
        });
    });
    Console.WriteLine("{0} has finished .....", guid);
    return output;
}

ArrayPool

public class ArrayPool<T>
{
    public int Size { get => pool.Count(); }
    public int maxSize = 3;
    public int circulingObjectCount = 0;
    private Queue<T> pool = new Queue<T>();
    private Func<T> constructorFunc;

    public ArrayPool(Func<T> constructorFunc) {
        this.constructorFunc = constructorFunc;
    }

    public Task Use(Func<T, Task> action)
    {
        T item = GetNextItem(); //DeQueue the item
        var t = action(item);
        t.ContinueWith(task => pool.Enqueue(item)); //Requeue the item
        return t;
    }

    private T GetNextItem()
    {
        //Create new object if pool is empty and not reached maxSize
        if (pool.Count == 0 && circulingObjectCount < maxSize)
        {
            T item = constructorFunc();
            circulingObjectCount++;
            Console.WriteLine("Pool empty, adding new item");
            return item;
        }
        //Wait for Queue to have at least 1 item
        WaitForReturns();

        return pool.Dequeue();
    }

    private void WaitForReturns()
    {
        long timeouts = 60000;
        while (pool.Count == 0 && timeouts > 0) { timeouts--; System.Threading.Thread.Sleep(1); }
        if(timeouts == 0)
        {
            throw new Exception("Wait timed-out");
        }
    }
}