管理多个 WebRequest 的更好方法

Question

我有一个组件在单独的线程中处理多个 Web 请求。每个 WebRequest 处理都是同步的。

public class WebRequestProcessor:System.ComponentModel.Component
{
    List<Worker> tlist = new List<Worker>();
    public void Start()
    {
        foreach(string url in urlList){
            // Create the thread object. This does not start the thread.
            Worker workerObject = new Worker();
            Thread workerThread = new Thread(workerObject.DoWork);

            // Start the worker thread.
            workerThread.Start(url);
            tlist.Add(workerThread);
        }
    }
}

public class Worker
{
    // This method will be called when the thread is started.
    public void DoWork(string url)
    {
        // prepare the web page we will be asking for
        HttpWebRequest  request  = (HttpWebRequest) 
            WebRequest.Create(url);

        // execute the request
        HttpWebResponse response = (HttpWebResponse)
            request.GetResponse();

        // we will read data via the response stream
        Stream resStream = response.GetResponseStream();

        // process stream
    }
}

现在我必须找到取消所有请求的最佳方法。

一种方法是将每个同步的 WebRequest 转换为 async 并使用 WebRequest.Abort 取消处理。

另一种方法是释放线程指针并允许所有线程使用 GC 死掉。

Answer 1

如果你想下载 1000 个文件，一次启动 1000 个线程肯定不是最好的选择。与一次只下载几个文件相比，它不仅可能不会为您带来任何加速，而且还需要至少 1 GB 的虚拟内存。创建线程的成本很高，尽量避免在循环中这样做。

您应该改为使用Parallel.ForEach() 以及请求和响应操作的异步版本。比如像这样（WPF代码）：

private void Start_Click(object sender, RoutedEventArgs e)
{
    m_tokenSource = new CancellationTokenSource();
    var urls = …;
    Task.Factory.StartNew(() => Start(urls, m_tokenSource.Token), m_tokenSource.Token);
}

private void Cancel_Click(object sender, RoutedEventArgs e)
{
    m_tokenSource.Cancel();
}

void Start(IEnumerable<string> urlList, CancellationToken token)
{
    Parallel.ForEach(urlList, new ParallelOptions { CancellationToken = token },
                     url => DownloadOne(url, token));

}

void DownloadOne(string url, CancellationToken token)
{
    ReportStart(url);

    try
    {
        var request = WebRequest.Create(url);

        var asyncResult = request.BeginGetResponse(null, null);

        WaitHandle.WaitAny(new[] { asyncResult.AsyncWaitHandle, token.WaitHandle });

        if (token.IsCancellationRequested)
        {
            request.Abort();
            return;
        }

        var response = request.EndGetResponse(asyncResult);

        using (var stream = response.GetResponseStream())
        {
            byte[] bytes = new byte[4096];

            while (true)
            {
                asyncResult = stream.BeginRead(bytes, 0, bytes.Length, null, null);

                WaitHandle.WaitAny(new[] { asyncResult.AsyncWaitHandle,
                                           token.WaitHandle });

                if (token.IsCancellationRequested)
                    break;

                var read = stream.EndRead(asyncResult);

                if (read == 0)
                    break;

                // do something with the downloaded bytes
            }
        }

        response.Close();
    }
    finally
    {
        ReportFinish(url);
    }
}

这样，当您取消操作时，所有下载都将被取消，并且不会开始新的下载。此外，您可能希望设置MaxDegreeOfParallelism 或ParallelOptions，这样您就不会一次进行太多下载。

我不确定你想对正在下载的文件做什么，所以使用StreamReader 可能是更好的选择。

Answer 2

我认为最好的解决方案是“Parallel Foreach Cancellation”。请检查以下代码。

1. 要实现取消，首先创建CancellationTokenSource 并将其传递给Parallel.ForEach 到option。
2. 如需取消，可致电CancellationTokenSource.Cancel()
3. 取消后会出现OperationCanceledException，需要处理。

有一篇关于Parallel Programming的好文章与我的回答相关，即Task Parallel Library By Sacha Barber on CodeProject。

CancellationTokenSource tokenSource = new CancellationTokenSource();
ParallelOptions options = new ParallelOptions()
{
    CancellationToken = tokenSource.Token
};

List<string> urlList = null;
//parallel foreach cancellation
try
{
    ParallelLoopResult result = Parallel.ForEach(urlList, options, (url) =>
    {
        // Create the thread object. This does not start the thread.
        Worker workerObject = new Worker();
        workerObject.DoWork(url);
    });
}
catch (OperationCanceledException ex)
{
    Console.WriteLine("Operation Cancelled");
}

---

更新

以下代码为“Parallel Foreach Cancellation Sample Code”。

class Program
{
    static void Main(string[] args)
    {
        List<int> data = ParallelEnumerable.Range(1, 10000).ToList();

        CancellationTokenSource tokenSource = new CancellationTokenSource();

        Task cancelTask = Task.Factory.StartNew(() =>
            {
                Thread.Sleep(1000);
                tokenSource.Cancel();
            });


        ParallelOptions options = new ParallelOptions()
        {
            CancellationToken = tokenSource.Token
        };


        //parallel foreach cancellation
        try
        {
            Parallel.ForEach(data,options, (x, state) =>
            {
                Console.WriteLine(x);
                Thread.Sleep(100);
            });
        }
        catch (OperationCanceledException ex)
        {
            Console.WriteLine("Operation Cancelled");
        }


        Console.ReadLine();
    }
}