azure - 如何在Azure WebJob函数中执行异步

Question

我有一个从服务器获取 api 数据的异步方法。当我在本地计算机上的控制台应用程序中运行此代码时，它会高速执行，每分钟在异步函数中推送数百个 http 调用。然而，当我将相同的代码从 Azure WebJob 队列消息触发时，它似乎同步运行并且我的数字在爬行 - 我确信我的方法中缺少一些简单的东西 - 感谢任何帮助。

(1) .. WebJob 函数监听队列上的消息并在收到消息时启动 api 获取进程:

public class Functions
    {
        // This function will get triggered/executed when a new message is written 
        // on an Azure Queue called queue.

        public static async Task ProcessQueueMessage ([QueueTrigger("myqueue")] string message, TextWriter log)
        {
            var getAPIData = new GetData();
            getAPIData.DoIt(message).Wait();
            log.WriteLine("*** done: " + message);
        }
    }

(2) azure 外部以异步模式快速工作的类...

 class GetData
    {
        // wrapper that is called by the message function trigger
        public async Task DoIt(string MessageFile)
        {
            await CallAPI(MessageFile);
        }

        public async Task<string> CallAPI(string MessageFile)
        {
            /// create a list of sample APIs to call...
            var apiCallList = new List<string>();
            apiCallList.Add("localhost/?q=1");
            apiCallList.Add("localhost/?q=2");
            apiCallList.Add("localhost/?q=3");
            apiCallList.Add("localhost/?q=4");
            apiCallList.Add("localhost/?q=5");

            // setup httpclient
            HttpClient client =
                new HttpClient() { MaxResponseContentBufferSize = 10000000 };
            var timeout = new TimeSpan(0, 5, 0); // 5 min timeout
            client.Timeout = timeout;

            // create a list of http api get Task...
            IEnumerable<Task<string>> allResults = apiCallList.Select(str => ProcessURLPageAsync(str, client));
            // wait for them all to complete, then move on...
            await Task.WhenAll(allResults);

            return allResults.ToString();
        }

        async Task<string> ProcessURLPageAsync(string APIAddressString, HttpClient client)
        {
            string page = "";
            HttpResponseMessage resX;

            try
            {
                // set the address to call
                Uri URL = new Uri(APIAddressString);
                // execute the call
                resX = await client.GetAsync(URL);
                page = await resX.Content.ReadAsStringAsync();
                string rslt = page;
                // do something with the api response data
            }
            catch (Exception ex)
            {
                // log error
            }
            return page;
        }

    }

Answer 1

首先，因为您的触发函数是异步，所以您应该使用await而不是.Wait()。等待会阻塞当前线程。

public static async Task ProcessQueueMessage([QueueTrigger("myqueue")] string message, TextWriter log)
{
    var getAPIData = new GetData();
    await getAPIData.DoIt(message);
    log.WriteLine("*** done: " + message);
}

无论如何，您都可以从documentation找到有用的信息。

Parallel execution

If you have multiple functions listening on different queues, the SDK will call them in parallel when messages are received simultaneously.

The same is true when multiple messages are received for a single queue. By default, the SDK gets a batch of 16 queue messages at a time and executes the function that processes them in parallel. The batch size is configurable. When the number being processed gets down to half of the batch size, the SDK gets another batch and starts processing those messages. Therefore the maximum number of concurrent messages being processed per function is one and a half times the batch size. This limit applies separately to each function that has a QueueTrigger attribute.

以下是配置批量大小的示例代码:

var config = new JobHostConfiguration();
config.Queues.BatchSize = 50;
var host = new JobHost(config);
host.RunAndBlock();

但是，同时运行太多线程并不总是一个好的选择，并且可能会导致性能不佳。

另一个选择是扩展您的网络作业:

Multiple instances

if your web app runs on multiple instances, a continuous WebJob runs on each machine, and each machine will wait for triggers and attempt to run functions. The WebJobs SDK queue trigger automatically prevents a function from processing a queue message multiple times; functions do not have to be written to be idempotent. However, if you want to ensure that only one instance of a function runs even when there are multiple instances of the host web app, you can use the Singleton attribute.