0

我需要从外部系统检索多个对象。外部系统支持多个同时请求（即线程），但可能会淹没外部系统 - 因此我希望能够异步检索多个对象，但我希望能够限制同时异步请求的数量。即我需要检索 100 个项目，但不想一次检索超过 25 个项目。当 25 的每个请求完成时，我想触发另一个检索，一旦它们全部完成，我想按请求的顺序返回所有结果（即在返回整个调用之前返回结果是没有意义的）。这种事情有什么推荐的模式吗？

这样的事情是否合适（显然是伪代码）？

  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);
   }
4

2 回答 2

1

将要处理的项目列表视为一个队列，其中 25 个处理线程将任务从队列中取出，处理一个任务，添加结果，然后重复直到队列为空：

 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();
        }
      }
    }
  }

你可以使用异步回调来做同样的事情，不需要使用线程。

于 2010-04-09T05:50:14.017 回答
0

拥有一些类似队列的结构来保存待处理的请求是一种非常常见的模式。在可能有多个处理层的 Web 应用程序中，您会看到一种“漏斗”式方法，处理更改的早期部分具有更大的队列。也可能对队列应用某种优先级，更高优先级的请求被打乱到队列的顶部。

在您的解决方案中要考虑的一件重要事情是，如果请求到达率高于您的处理率（这可能是由于拒绝服务攻击，或者只是今天的某些处理异常缓慢），那么您的队列将会增加无界。您需要制定一些策略，例如当队列深度超过某个值时立即拒绝新请求。

于 2010-04-09T07:09:16.213 回答