如何在 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

首先因为你的触发函数是async，你应该使用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 找到有用的信息

并行执行

如果您有多个函数监听不同的队列，SDK 会在同时接收消息时并行调用它们。

当一个队列接收到多条消息时也是如此。默认情况下，SDK 一次获取一批 16 条队列消息并执行并行处理它们的函数。 The batch size is configurable。当正在处理的数量减少到批处理大小的一半时，SDK 会获取另一个批处理并开始处理这些消息。 因此，每个函数处理的最大并发消息数是批量大小的一倍半。此限制分别适用于具有 QueueTrigger 属性的每个函数。

这是一个配置批量大小的示例代码：

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

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

另一种选择是扩展您的网络作业：

多个实例

如果您的 Web 应用程序在多个实例上运行，则每台机器上都会运行一个连续的 WebJob，并且每台机器都会等待触发器并尝试运行函数。 WebJobs SDK 队列触发器自动防止函数多次处理队列消息；函数不必写成幂等的。但是，如果您想确保即使在宿主 Web 应用程序有多个实例的情况下也只有一个函数实例运行，您可以使用 Singleton 属性。

Answer 2

阅读此Webjobs SDK documentation - 您应该期望的行为是您的进程将运行并一次处理一条消息，但如果创建更多实例（您的应用程序服务）将扩大规模。如果您有多个队列，它们将并行触发。

为了提高性能，请看我发给你的链接中的配置设置部分，它是指可以批量触发的消息数量。

如果你想并行处理多条消息，并且不想依赖实例扩展，那么你需要使用线程来代替（异步不是多线程并行，而是更有效地利用您正在使用的线程）。因此，您的队列触发函数应该从队列中读取消息，创建一个线程并“触发并忘记”该线程，然后从触发函数返回。这会将消息标记为已处理，并允许处理队列中的下一条消息，即使理论上您仍在处理较早的消息。请注意，您需要包含自己的错误处理逻辑，并确保如果您的线程抛出异常或无法处理消息（例如，将其放入毒队列），数据不会丢失。

另一种选择是不使用 [queuetrigger] 属性，而是直接使用 Azure 存储队列 sdk API 函数根据您的要求连接和处理消息。