如何在不创建多个线程的情况下发出并发请求？

Question

有人可以展示如何在不创建多个线程的情况下发出并发请求吗？例如，我想要一个发出 100 个 Web 请求的程序，并且我不希望任何时候超过 8 个并发请求。我不想为 8 个并发请求创建 8 个线程。当一个线程发出异步请求时，同一个线程可以用于发出下一个请求，依此类推。很抱歉，但我无法解决这个问题，并希望看到最好的解决方案。如果不清楚，我正在谈论的请求是异步的。我希望看到一个不使用任何锁，并使用内置类来完成工作的解决方案。

这是我想出的一些代码，但它没有做它应该做的事情。

Task.Run(async () =>
                        {
                            var outstandingRequests = 0;
                            var requestCount = 0;
                            var tasks = new List<Task>(concurrentRequests);
                            while (requestCount < maxRequests)
                            {
                                if (outstandingRequests < concurrentRequests)
                                {
                                    tasks.Add(svc.GetDataAsync());  // a method that makes an async request
                                    Interlocked.Increment(ref outstandingRequests);
                                }
                                else
                                {
                                    var t = await Task.WhenAny(tasks);                                                                       
                                    Interlocked.Decrement(ref outstandingRequests);
                                    Interlocked.Increment(ref requestCount);
                                }
                            }
                            await Task.WhenAll(tasks);
                        }).Wait();

输出：

[] 1 Sending Request...Received Response 490,835.00 bytes in 15.6 sec
[] 2 Sending Request...
[] 3 Sending Request...
[] 4 Sending Request...
[] 5 Sending Request...
[] 6 Sending Request...
[] 7 Sending Request...
[] 8 Sending Request...
[] 9 Sending Request...

我已将concurrentRequests 设置为 5，因此上述代码中存在一些错误，因为它并行发出 8 个请求。最初它只并行发出 5 个请求，但一旦一个请求完成，它就又触发了 4 个请求（应该只触发一个）。

不得不修复一些错误，但现在一切都解决了：

Task.Run(async () =>
                        {
                            var outstandingRequests = 0;
                            var requestCount = 0;
                            // adding and removing from a List<> at the same time is not thread-safe,
                            // so have to use a SynchronizedCollection<>
                            var tasks = new SynchronizedCollection<Task>();
                            while (requestCount < maxRequests)
                            {
                                if (outstandingRequests < concurrentRequests)
                                {
                                    tasks.Add(svc.GetDataAsync(uri)); // this will be your method that makes async web call and returns a Task to signal completion of async call
                                    Interlocked.Increment(ref outstandingRequests);
                                    Interlocked.Increment(ref requestCount);
                                }
                                else
                                {                                    
                                    **tasks.Remove(await Task.WhenAny(tasks));**
                                    Interlocked.Decrement(ref outstandingRequests);                                    
                                }
                            }
                            await Task.WhenAll(tasks);
                        }).Wait();

如果有更好的方法，请告诉我。

Answer 1

看起来您正在尝试重新发明线程池。不要那样做 - 只需使用现有功能：http://msdn.microsoft.com/en-us/library/system.threading.threadpool.aspx
或者您可以使用异步版本的请求方法——它们也基于线程池。

Answer 2

这个怎么样：

Parallel.Invoke (new ParallelOptions { MaxDegreeOfParallelism = 8 },
    svcs.Select (svc => svc.GetDataAsync ()).ToArray ()) ;

有一个有限并发任务计划程序here 的示例 Microsoft 实现。请参阅 SO 问题 System.Threading.Tasks - Limit the number of concurrent Tasks 和 .Net TPL: Limited Concurrency Level Task scheduler with task priority?。