C# 异步 Web 请求 : Perform Action When All Requests Are Completed

Question

我在 C# 中有这个基本的抓取控制台应用程序，它异步使用 WebRequest 从站点列表中获取 html。它工作正常，但我如何设置一个触发器，在处理完列表中的每个站点后触发？

我花了几个小时在线研究各种解决方案，包括 MS 文档，但没有一个通过代码提供直接的答案。我读过有关 IAsyncResult.AsyncWaitHandle 的信息，但不知道如何将它集成到我的代码中。我只想在所有线程完成处理或超时时调用自定义函数。

一个技巧是我永远不知道我的列表中有多少网站(这是用户定义的)，所以我需要一个足够强大的解决方案来等待 5 个事件完成 100,000 个事件。

谢谢。工作代码如下:

using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Net;
using System.Threading;


namespace AsyncApp_01
{
    class Program
    {
        static void Main(string[] args)
        {
            ArrayList alSites = new ArrayList();
            alSites.Add("http://www.google.com");
            alSites.Add("http://www.lostspires.com");

            ScanSites(alSites);

            Console.Read();
        }

        private static void ScanSites(ArrayList sites)
        {
            foreach (string uriString in sites)
            {
                WebRequest request = HttpWebRequest.Create(uriString);
                request.Method = "GET";
                object data = new object(); //container for our "Stuff"

                // RequestState is a custom class to pass info to the callback
                RequestState state = new RequestState(request, data, uriString);
                IAsyncResult result = request.BeginGetResponse(new AsyncCallback(UpdateItem), state);


                //Register the timeout callback
                ThreadPool.RegisterWaitForSingleObject(result.AsyncWaitHandle, new WaitOrTimerCallback(ScanTimeoutCallback), state, (30 * 1000), true);

            }
        }


        private static void UpdateItem(IAsyncResult result)
        {
            // grab the custom state object
            RequestState state = (RequestState)result.AsyncState;
            WebRequest request = (WebRequest)state.Request;

            // get the Response
            HttpWebResponse response = (HttpWebResponse)request.EndGetResponse(result);
            Stream s = (Stream)response.GetResponseStream();
            StreamReader readStream = new StreamReader(s);

            // dataString will hold the entire contents of the requested page if we need it.
            string dataString = readStream.ReadToEnd();
            response.Close();
            s.Close();
            readStream.Close();

            Console.WriteLine(dataString);
        }


        private static void ScanTimeoutCallback(object state, bool timedOut)
        {
            if (timedOut)
            {
                RequestState reqState = (RequestState)state;
                if (reqState != null)
                {
                    reqState.Request.Abort();
                }
                Console.WriteLine("aborted- timeout");
            }
        } 


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

        }
    }
}

任何人都可以告诉我如何限制并发线程数的奖励积分。例如，如果我有 100 个站点要处理，我该如何设置以便一次处理 10 个站点，但不能更多。我不想打开 100 个线程。

Answer 1

这是我拼凑的一个快速示例。我删除了 WebClient 实现，因为您似乎正在使用 WebRequest 实现。我也在使用 .Net 4 的 ConcurrentBag:

public class Scraper
{
    private readonly IEnumerable<string> _sites;
    private readonly ConcurrentBag<string> _data;
    private volatile int _count;
    private readonly int _total;
    public Scraper(IEnumerable<string> sites)
    {
        _sites = sites;
        _data = new ConcurrentBag<string>();
        _total = sites.Count();
    }

    public void Start()
    {
        foreach (var site in _sites)
        {
            ScrapeSite(site);
        }
    }

    private void ScrapeSite(string site)
    {
        var req = WebRequest.Create(site);
        req.BeginGetResponse(AsyncCallback, req);
    }

    private void AsyncCallback(IAsyncResult ar)
    {
        Interlocked.Increment(ref _count);
        var req = ar.AsyncState as WebRequest;

        var result = req.EndGetResponse(ar);
        var reader = new StreamReader(result.GetResponseStream());
        var data = reader.ReadToEnd();
        this.OnSiteScraped(req.RequestUri.AbsoluteUri, data);
        _data.Add(data);
        if (_count == _total)
        {
            OnScrapingComplete();
        }
    }

    private void OnSiteScraped(string site, string data)
    {
        var handler = this.SiteScraped;
        if (handler != null)
        {
            handler(this, new SiteScrapedEventArgs(site, data));
        }
    }

    private void OnScrapingComplete()
    {
        var handler = this.ScrapingComplete;
        if (handler != null)
        {
            handler(this, new ScrapingCompletedEventArgs(_data));
        }
    }

    public event EventHandler<SiteScrapedEventArgs> SiteScraped;
    public event EventHandler<ScrapingCompletedEventArgs> ScrapingComplete;
}

public class SiteScrapedEventArgs : EventArgs
{
    public string Site { get; private set; }
    public string Data { get; private set; }
    public SiteScrapedEventArgs(string site, string data)
    {
        this.Site = site;
        this.Data = data;
    }
}

好的，我创建了一些基本类，这应该可以解决问题。如果这还不够，很抱歉，我帮不了您:

 public class RankedPage
    {
        public int Rank { get; set; }
        public string Site { get; set; }
    }

    public class WebRequestData
    {
        public WebRequest WebRequest { get; set; }
        public RankedPage Page { get; set; }
    }

    public class Scraper
    {
        private readonly IEnumerable<RankedPage> _sites;
        private readonly ConcurrentBag<KeyValuePair<RankedPage,string>> _data;
        private volatile int _count;
        private readonly int _total;
        public Scraper(IEnumerable<RankedPage> sites)
        {
            _sites = sites;
            _data = new ConcurrentBag<KeyValuePair<RankedPage, string>>();
            _total = sites.Count();
        }

        public void Start()
        {
            foreach (var site in _sites)
            {
                ScrapeSite(site);
            }
        }

        private void ScrapeSite(RankedPage site)
        {
            var req = WebRequest.Create(site.Site);
            req.BeginGetResponse(AsyncCallback, new WebRequestData{ Page = site, WebRequest = req});
        }

        private void AsyncCallback(IAsyncResult ar)
        {
            Interlocked.Increment(ref _count);
            var webRequestData = ar.AsyncState as WebRequestData;

            var req = webRequestData.WebRequest;
            var result = req.EndGetResponse(ar);
            var reader = new StreamReader(result.GetResponseStream());
            var data = reader.ReadToEnd();
            this.OnSiteScraped(webRequestData.Page, data);
            _data.Add(new KeyValuePair<RankedPage, string>(webRequestData.Page,data));
            if (_count == _total)
            {
                OnScrapingComplete();
            }
        }

        private void OnSiteScraped(RankedPage page, string data)
        {
            var handler = this.SiteScraped;
            if (handler != null)
            {
                handler(this, new SiteScrapedEventArgs(page, data));
            }
        }

        private void OnScrapingComplete()
        {
            var handler = this.ScrapingComplete;
            if (handler != null)
            {
                handler(this, new ScrapingCompletedEventArgs(_data));
            }
        }

        public event EventHandler<SiteScrapedEventArgs> SiteScraped;
        public event EventHandler<ScrapingCompletedEventArgs> ScrapingComplete;
    }

    public class SiteScrapedEventArgs : EventArgs
    {
        public RankedPage Site { get; private set; }
        public string Data { get; private set; }
        public SiteScrapedEventArgs(RankedPage site, string data)
        {
            this.Site = site;
            this.Data = data;
        }
    }

    public class ScrapingCompletedEventArgs : EventArgs
    {
        public IEnumerable<KeyValuePair<RankedPage,string >> SiteData { get; private set; }
        public ScrapingCompletedEventArgs(IEnumerable<KeyValuePair<RankedPage, string>> siteData)
        {
            this.SiteData = siteData;
        }
    }