Padrão para limitar o número de chamadas assíncronas simultâneas

Question

Preciso recuperar vários objetos de um sistema externo.O sistema externo suporta múltiplas solicitações simultâneas (ou seja,threads), mas é possível inundar o sistema externo - portanto, quero poder recuperar vários objetos de forma assíncrona, mas quero ser capaz de limitar o número de solicitações assíncronas simultâneas.ou sejaPreciso recuperar 100 itens, mas não quero ser recuperando mais de 25 deles ao mesmo tempo.Quando cada solicitação das 25 for concluída, quero acionar outra recuperação e, quando todas forem concluídas, quero retornar todos os resultados na ordem em que foram solicitados (ou seja,não faz sentido retornar os resultados até que toda a chamada seja retornada).Existem padrões recomendados para esse tipo de coisa?

Algo assim seria apropriado (pseudocódigo, obviamente)?

  private List<externalSystemObjects> returnedObjects = new List<externalSystemObjects>;

  public List<externalSystemObjects> GetObjects(List<string> ids)
  {
      int callCount = 0;
      int maxCallCount = 25;
      WaitHandle[] handles;

      foreach(id in itemIds to get)
      {
          if(callCount < maxCallCount)
          {
               WaitHandle handle = executeCall(id, callback);
               addWaitHandleToWaitArray(handle)
          }
      else
      {
           int returnedCallId = WaitHandle.WaitAny(handles);
           removeReturnedCallFromWaitHandles(handles);
      }
   }

   WaitHandle.WaitAll(handles);

   return returnedObjects;
   }

   public void callback(object result)
   {
         returnedObjects.Add(result);
   }

Answer 1

Considere a lista de itens a serem processados ​​como uma fila da qual 25 threads de processamento desenfileiram tarefas, processam uma tarefa, adicionam o resultado e repetem até que a fila esteja vazia:

 class Program
  {
    class State
    {
      public EventWaitHandle Done;
      public int runningThreads;
      public List<string> itemsToProcess;
      public List<string> itemsResponses;
    }

    static void Main(string[] args)
    {
      State state = new State();

      state.itemsResponses = new List<string>(1000);
      state.itemsToProcess = new List<string>(1000);
      for (int i = 0; i < 1000; ++i)
      {
        state.itemsToProcess.Add(String.Format("Request {0}", i));
      }

      state.runningThreads = 25;
      state.Done = new AutoResetEvent(false);

      for (int i = 0; i < 25; ++i)
      {
        Thread t =new Thread(new ParameterizedThreadStart(Processing));
        t.Start(state);
      }

      state.Done.WaitOne();

      foreach (string s in state.itemsResponses)
      {
        Console.WriteLine("{0}", s);
      }
    }

    private static void Processing(object param)
    {
      Debug.Assert(param is State);
      State state = param as State;

      try
      {
        do
        {
          string item = null;
          lock (state.itemsToProcess)
          {
            if (state.itemsToProcess.Count > 0)
            {
              item = state.itemsToProcess[0];
              state.itemsToProcess.RemoveAt(0);
            }
          }
          if (null == item)
          {
            break;
          }
          // Simulate some processing
          Thread.Sleep(10);
          string response = String.Format("Response for {0} on thread: {1}", item, Thread.CurrentThread.ManagedThreadId);
          lock (state.itemsResponses)
          {
            state.itemsResponses.Add(response);
          }
        } while (true);

      }
      catch (Exception)
      {
        // ...
      }
      finally
      {
        int threadsLeft = Interlocked.Decrement(ref state.runningThreads);
        if (0 == threadsLeft)
        {
          state.Done.Set();
        }
      }
    }
  }

Você pode fazer o mesmo usando retornos de chamada assíncronos, não há necessidade de usar threads.

Answer 2

Ter alguma estrutura semelhante à fila para manter as solicitações pendentes é um padrão bastante comum. Nos aplicativos da Web, onde pode haver várias camadas de processamento, você vê uma abordagem de estilo "funil" com as partes iniciais da mudança de processamento com filas maiores. Também pode haver algum tipo de priorização aplicada a filas, sendo mais altas solicitações de prioridade sendo embaralhadas no topo da fila.

Uma coisa importante a considerar em sua solução é que, se a taxa de chegada da solicitação for maior que a sua taxa de processamento (isso pode ser devido a um ataque de negação de serviço, ou apenas que parte do processamento é incomumente lenta hoje), então suas filas aumentarão sem limite. Você precisa ter alguma política que recuse novas solicitações imediatamente quando a profundidade da fila exceder algum valor.