限制Azure Functions队列上的并发作业数

Question

我在Azure中有一个Function项目,当项目放入队列时会触发该应用程序.它看起来像这样(大大简化):

public static async Task Run(string myQueueItem, TraceWriter log)
{
    using (var client = new HttpClient())
    {
        client.BaseAddress = new Uri(Config.APIUri);
        client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));

        StringContent httpContent = new StringContent(myQueueItem, Encoding.UTF8, "application/json");
        HttpResponseMessage response = await client.PostAsync("/api/devices/data", httpContent);
        response.EnsureSuccessStatusCode();

        string json = await response.Content.ReadAsStringAsync();
        ApiResponse apiResponse = JsonConvert.DeserializeObject<ApiResponse>(json);

        log.Info($"Activity data successfully sent to platform in {apiResponse.elapsed}ms.  Tracking number: {apiResponse.tracking}");
    }
}

这一切都很好,运行得很好.每次将项目放入队列时,我们都会将数据发送到我们这边的某个API并记录响应.凉.

当"产生队列消息的东西"出现大幅增加并且许多项目立即被放入队列时,就会出现问题.这往往会在一分钟内发生大约1,000到1,500件物品.错误日志将具有以下内容:

2017-02-14T01:45:31.692 mscorlib:执行函数时出现异常:Functions.SendToLimeade.f-SendToLimeade __- 1078179529:发送请求时发生错误.系统:无法连接到远程服务器.系统:通常只允许使用每个套接字地址(协议/网络地址/端口)123.123.123.123:443.

起初,我认为这是Azure功能应用程序运行本地套接字的问题,如此处所示.但是,我注意到了IP地址.IP地址123.123.123.123(当然在本例中已更改)是我们的IP地址,即HttpClient发布的IP地址.所以,现在我想知道是不是我们的服务器用完了套接字来处理这些请求.

无论哪种方式,我们都会遇到扩展问题.我正试图找出解决问题的最佳方法.

一些想法:

1. 如果它是本地套接字限制,则上面的文章中有一个使用增加本地端口范围的示例Req.ServicePoint.BindIPEndPointDelegate.这似乎很有希望,但是当你真正需要扩展时,你会怎么做？我不希望这个问题在2年内回归.
2. 如果它是一个远程限制,看起来我可以控制函数运行时将一次处理多少消息.这里有一篇有趣的文章说你可以设置serviceBus.maxConcurrentCalls为1,一次只处理一条消息.也许我可以将其设置为相对较低的数字.现在,在某些时候,我们的队列填充速度将超过我们处理它们的速度,但此时答案是在我们的末端添加更多服务器.
3. 多个Azure功能应用程序？如果我有多个Azure Functions应用程序并且它们都在同一队列上触发,会发生什么？Azure是否足够聪明,可以在功能应用程序之间分配工作,我可以让大量机器处理我的队列,可以根据需要按比例放大或缩小？
4. 我也遇到了保持活力.在我看来,如果我可以以某种方式保持我的套接字打开,因为队列消息泛滥,它可能会有很大帮助.这是可能的,以及关于我如何做到这一点的任何提示？

对于此类系统的推荐(可扩展!)设计的任何见解将不胜感激!

Answer 1

我想我已经找到了解决方案.我过去一直在运行这些变化3小时6个小时,我有零插槽错误.在我每30分钟左右大批量地收到这些错误之前.

首先,我添加了一个新类来管理HttpClient.

public static class Connection
{
    public static HttpClient Client { get; private set; }

    static Connection()
    {
        Client = new HttpClient();

        Client.BaseAddress = new Uri(Config.APIUri);
        Client.DefaultRequestHeaders.Add("Connection", "Keep-Alive");
        Client.DefaultRequestHeaders.Add("Keep-Alive", "timeout=600");
        Client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
    }
}

现在,我们有一个静态实例HttpClient,我们用于每次调用函数.根据我的研究,强烈建议尽可能长时间地保持HttpClient实例,一切都是线程安全的,HttpClient会将请求排队并优化对同一主机的请求.注意我也设置了Keep-Alive标题(我认为这是默认值,但我认为我会隐含).

在我的函数中,我只是抓住静态HttpClient实例,如:

var client = Connection.Client;
StringContent httpContent = new StringContent(myQueueItem, Encoding.UTF8, "application/json");
HttpResponseMessage response = await client.PostAsync("/api/devices/data", httpContent);
response.EnsureSuccessStatusCode();

我还没有真正深入分析套接字级别发生的事情(我不得不问问我们的IT人员是否能够在负载均衡器上看到这种流量),但我希望它只是保持单个套接字对我们的服务器开放,并在处理队列项时进行一堆HTTP调用.无论如何,无论它做什么似乎都在起作用.也许有人对如何改进有一些想法.

Answer 2

我认为代码错误是因为： using (var client = new HttpClient())

引用自不正确的实例化反模式：

这种技术不可扩展。为每个用户请求创建一个新的 HttpClient 对象。在高负载下，Web 服务器可能会耗尽可用套接字的数量。

Answer 3

如果您在专用 Web 应用程序上使用消耗计划而不是函数，则#3 或多或少会立即发生。函数将检测到您有一个很大的消息队列，并将添加实例，直到队列长度稳定。

maxConcurrentCalls仅适用于每个实例，允许您限制每个实例的并发数。基本上，您的处理速率是maxConcurrentCalls * instanceCount。

控制全局吞吐量的唯一方法是在您选择的大小的专用 Web 应用程序上使用函数。每个应用程序都会轮询队列并根据需要获取工作。

最佳的扩展解决方案将改善 123.123.123.123 上的负载平衡，以便它可以处理来自向上/向下扩展的函数的任意数量的请求，以满足队列压力。

保持活动状态对于持久连接很有用，但函数执行不被视为持久连接。将来，我们将尝试向 Functions 添加“自带绑定”，这样您就可以根据需要实现连接池。