C# 使用异步加速并行 Web 请求

Question

所以我有这个代码： 这是 main 函数，一个并行的 for 循环，遍历所有需要发布的数据并调用函数

 ParallelOptions pOpt = new ParallelOptions();
    pOpt.MaxDegreeOfParallelism = 30;
    Parallel.For(0, maxsize, pOpt, (index,loopstate) => {

                    //Calls the function where all the webrequests are made
                    CallRequests(data1,data2);

                    if (isAborted)
                        loopstate.Stop();
                });

这个函数在并行循环中被调用

public static void CallRequests(string data1, string data2)     
    {
        var cookie = new CookieContainer();
        var postData =  Parameters[23] + data1 +
                        Parameters[24] + data2;

        HttpWebRequest getRequest = (HttpWebRequest)WebRequest.Create(Parameters[25]);
        getRequest.Accept = Parameters[26];
        getRequest.KeepAlive = true;
        getRequest.Referer = Parameters[27];
        getRequest.CookieContainer = cookie;
        getRequest.UserAgent = Parameters[28];
        getRequest.Method = WebRequestMethods.Http.Post;
        getRequest.AllowWriteStreamBuffering = true;
        getRequest.ProtocolVersion = HttpVersion.Version10;
        getRequest.AllowAutoRedirect = false;
        getRequest.ContentType = Parameters[29];
        getRequest.ReadWriteTimeout = 5000;
        getRequest.Timeout = 5000;
        getRequest.Proxy = null;

        byte[] byteArray = Encoding.ASCII.GetBytes(postData);
        getRequest.ContentLength = byteArray.Length;
        Stream newStream = getRequest.GetRequestStream(); //open connection
        newStream.Write(byteArray, 0, byteArray.Length); // Send the data.
        newStream.Close();

        HttpWebResponse getResponse = (HttpWebResponse)getRequest.GetResponse();

        if (getResponse.Headers["Location"] == Parameters[30])
        {
            //These are simple get requests to retrieve the source code using the same format as above.
            //I need to preserve the cookie
            GetRequets(data1, data2, Parameters[31], Parameters[13], cookie);
            GetRequets(data1, data2, Parameters[32], Parameters[15], cookie);
        }
    }

从我所见和被告知的情况来看，我知道使这些请求异步比使用并行循环更好。我的方法在处理器上也很重。我想知道如何使这些请求异步，但也保留多线程方面。在 post 请求完成后，我还需要保留 cookie。

Answer 1

将CallRequests 方法转换为async 实际上只是将同步方法调用切换为带有await 关键字的异步方法调用，并将方法签名更改为返回Task。

类似这样的：

public static async Task CallRequestsAsync(string data1, string data2)
{
    var cookie = new CookieContainer();
    var postData = Parameters[23] + data1 +
                    Parameters[24] + data2;

    HttpWebRequest getRequest = (HttpWebRequest)WebRequest.Create(Parameters[25]);
    getRequest.Accept = Parameters[26];
    getRequest.KeepAlive = true;
    getRequest.Referer = Parameters[27];
    getRequest.CookieContainer = cookie;
    getRequest.UserAgent = Parameters[28];
    getRequest.Method = WebRequestMethods.Http.Post;
    getRequest.AllowWriteStreamBuffering = true;
    getRequest.ProtocolVersion = HttpVersion.Version10;
    getRequest.AllowAutoRedirect = false;
    getRequest.ContentType = Parameters[29];
    getRequest.ReadWriteTimeout = 5000;
    getRequest.Timeout = 5000;
    getRequest.Proxy = null;

    byte[] byteArray = Encoding.ASCII.GetBytes(postData);
    getRequest.ContentLength = byteArray.Length;
    Stream newStream =await getRequest.GetRequestStreamAsync(); //open connection
    await newStream.WriteAsync(byteArray, 0, byteArray.Length); // Send the data.
    newStream.Close();

    HttpWebResponse getResponse = (HttpWebResponse)getRequest.GetResponse();

    if (getResponse.Headers["Location"] == Parameters[30])
    {
        //These are simple get requests to retrieve the source code using the same format as above.
        //I need to preserve the cookie
        GetRequets(data1, data2, Parameters[31], Parameters[13], cookie);
        GetRequets(data1, data2, Parameters[32], Parameters[15], cookie);
    }
}

但是，这本身并不能真正让您到达任何地方，因为您仍然需要在 main 方法中等待返回的任务。一个非常直接（如果有点生硬）的方法是简单地调用Task.WaitAll()（如果调用方法本身要变为异步，则调用await Task.WhenAll()）。像这样的：

var tasks = Enumerable.Range(0, maxsize).Select(index => CallRequestsAsync(data1, data2));
Task.WaitAll(tasks.ToArray());

但是，这真的很生硬，并且无法控制并行运行的迭代次数等。我更喜欢使用TPL dataflow library 来处理这类事情。这个库提供了一种并行链接异步（或同步）操作并将它们从一个“处理块”传递到下一个的方法。它有无数用于调整并行度、缓冲区大小等的选项。

详细的公开超出了此答案的可能范围，因此我鼓励您阅读它，但一种可能的方法是将其简单地推送到操作块 - 如下所示：

var actionBlock = new ActionBlock<int>(async index =>
{
    await CallRequestsAsync(data1, data2);
}, new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 30,
    BoundedCapacity = 100,
});
for (int i=0; i <= maxsize; i++)
{
    actionBlock.Post(i); // or await actionBlock.SendAsync(i) if calling method is also async
}

actionBlock.Complete();
actionBlock.Completion.Wait(); // or await actionBlock.Completion if calling method is also async

我应该顺便提一下我的答案范围之外的几个附加点：

1. 它看起来就像你的CallRequests 方法正在用它的结果更新一些外部变量。在可能的情况下，最好避免这种模式，并让方法 return 稍后进行整理（TPL 数据流库通过TransformBlock<> 处理）。如果更新外部状态是不可避免的，那么请确保您已经考虑了我的回答范围之外的多线程影响（死锁、竞争条件等）。
2. 我假设index 的一些有用属性在您为问题创建最小描述时丢失了？它是否索引到参数列表或类似的东西？如果是这样，您始终可以直接迭代这些并将ActionBlock<int> 更改为ActionBlock<{--whatever the type of your parameter is--}>
3. 确保您了解多线程/并行执行与异步之间的区别。肯定有一些相似之处/重叠，但仅仅制作异步并不能使其成为多线程，反之亦然。