如何在 C# 中将队列任务添加到线程池

Question

我对这个线程的东西还是个新手。假设我有 50000 个 URL，我想同时获取这些 URL 的内容，比如每 10 个 URL 一起处理。然后一旦这些 URL 之一完成处理，程序应该从队列列表中添加另一个 1，直到它完成处理列表中的所有 URL。现在我怎么能用 C# 做到这一点.. 这是我目前正在做的代码..

 class RequestState
        {
            public WebRequest Request;

        // holds the request 
        public object Data;

        // store any data in this 
        public string SiteUrl;

        // holds the UrlString to match up results (Database lookup, etc). 

        public RequestState(WebRequest request, object data, string siteUrl)
        {
            this.Request = request;
            this.Data = data;
            this.SiteUrl = siteUrl;
        }
    }

    private void PROCESS_URLS_Click(object sender, EventArgs e)
    {
        //run the process
        process_URLs();
    }

private int ThreadsCount = 0;

  private void process_URLs()
    {
       //count threads number
        ThreadsCount = URLS_LISTVIEW.SelectedItems.Count;

       //loop through all URLs in listview
        for (int i = 0; i < URLS_LISTVIEW.SelectedItems.Count; i++)
        {
            try
            {
                //get url string
                string myURLs = URLS_LISTVIEW.SelectedItems[i].SubItems[0].Text.Trim();

                // for each URL in the collection...
                WebRequest request = HttpWebRequest.Create(myURLs);
                request.Method = "GET";
                object data = new object();

                RequestState state = new RequestState(request, data, myURLs);
                IAsyncResult result = request.BeginGetResponse(new AsyncCallback(UpdateItem), state);
                ThreadPool.RegisterWaitForSingleObject(result.AsyncWaitHandle, new WaitOrTimerCallback(ScanTimeoutCallback), state, (30 * 1000), true);

            }
            catch (ThreadStateException es)
            {
                MessageBox.Show(es.Message);
            }

        }



    }




 private void UpdateItem(IAsyncResult result)
    {
        RequestState state = (RequestState)result.AsyncState;
        WebRequest request = (WebRequest)state.Request;
        try
        {// grab the custom state object
            // get the Response
            HttpWebResponse response = (HttpWebResponse)request.EndGetResponse(result);

            // process the response...
            Stream s = (Stream)response.GetResponseStream();
            StreamReader readStream = new StreamReader(s);

            //data grabbed
            string dataString = readStream.ReadToEnd();
            response.Close();
            s.Close();
            readStream.Close();



        //finished grabbing content for this thread.
        ThreadsCount = ThreadsCount - 1;


        //if all threads finished running then execute final code to tell the user the process finished
        if (ThreadsCount < 1)
        {
            //show message
            MessageBox.Show("finished");
        }

       // Thread.Sleep(400);

    }





private static void ScanTimeoutCallback(object state, bool timedOut)
    {
        if (timedOut)
        {
            RequestState reqState = (RequestState)state;

            if (reqState != null)
                reqState.Request.Abort();


        }
    }

任何想法都将不胜感激:)

亲切的问候，

Answer 1

看看 TPL，有一个选项可以指定最大并行度：

List<string> UriList = new List<string>();
...
Parallel.ForEach(UriList, 
                 new ParallelOptions() {MaxDegreeOfParallelism=10}, 
                 (x) =>
{
    ProcessUrl(x);
});

这将最多并行处理 10 个 Url，因为我们使用 Parallel.Foreach() 的重载允许我们指定 MaxDegreeOfParallelism。

编辑：

这里有一个简单的例子，它从http://google.com 并行下载 Html 50 次（但最多同时有 10 个线程）并将结果存储在一个数组中：

List<string> UriList = new List<string>();
for(int i =0;i<50;i++)
    UriList.Add("http://google.com");

string[] HtmlResults = new string[UriList.Count];

Parallel.ForEach(UriList, 
                 new ParallelOptions() { MaxDegreeOfParallelism = 10 }, 
                 (url, i, j) =>
{
    WebClient wc = new WebClient();
    HtmlResults[j] = wc.DownloadString(url);
});

不要造成更多混乱，但在您的特定情况下，PLINQ 也可以很好地工作，因为要处理的项目之间没有依赖关系，并且您的实际结果是 URL 被“转换”为：

var htmlResultList = UriList.AsParallel()
                            .WithDegreeOfParallelism(10)
                            .AsOrdered()
                            .Select(url => { WebClient wc = new WebClient(); return wc.DownloadString(url); })
                            .ToList();

Answer 2

（这应该是@BrokenGlass下的评论，但我还不能发布cmets）

您可以查看this article，了解如何使用并行处理和 PLINQ 来完成您的任务。前面的整套文章也有一些很好的信息。

编辑：如果这是独立的，请生成一个新线程以在后台运行此部分，以免导致 UI 无响应。

编辑 2：如果需要，您还可以将字符串放入 ConcurrentQueue，以便在查找项目时从 UI 添加项目。